Initial idea

.env.example

@@ -0,0 +1,17 @@
+SERVER_HOST=127.0.0.1
+SERVER_PORT=3000
+
+DATABASE_URL=postgres://postgres:postgres@localhost:5432/pds
+
+OBJECT_STORAGE_ENDPOINT=
+OBJECT_STORAGE_REGION=us-east-1
+OBJECT_STORAGE_BUCKET=pds-blobs
+OBJECT_STORAGE_ACCESS_KEY=
+OBJECT_STORAGE_SECRET_KEY=
+
+# Set to 'true' for MinIO or other services that need path-style addressing
+OBJECT_STORAGE_FORCE_PATH_STYLE=false
+
+JWT_SECRET=your-super-secret-jwt-key-please-change-me
+PDS_HOSTNAME=localhost:3000 # The public-facing hostname of the PDS
+PLC_URL=plc.directory

.gitignore

@@ -0,0 +1,6 @@
+/target
+src_old
+.sqlx
+
+.env
+

Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "bspds"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+anyhow = "1.0.100"
+axum = "0.8.7"
+bcrypt = "0.17.1"
+bytes = "1.11.0"
+chrono = { version = "0.4.42", features = ["serde"] }
+cid = "0.11.1"
+dotenvy = "0.15.7"
+jacquard = "0.9.3"
+jacquard-axum = "0.9.2"
+jacquard-repo = "0.9.2"
+jsonwebtoken = { version = "10.2.0", features = ["rust_crypto"] }
+multihash = "0.19.3"
+reqwest = { version = "0.12.24", features = ["json"] }
+serde = { version = "1.0.228", features = ["derive"] }
+serde_ipld_dagcbor = "0.6.4"
+serde_json = "1.0.145"
+sha2 = "0.10.9"
+sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "json"] }
+tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "time"] }
+tracing = "0.1.43"
+tracing-subscriber = "0.3.22"
+uuid = { version = "1.19.0", features = ["v4", "fast-rng"] }

TODO.md

@@ -0,0 +1,108 @@
+# Implementation TODOs
+
+Lewis' special big boy todofile
+
+## 1. Server Infrastructure & Health
+- [x] Health Check
+    - [x] Implement `GET /health` endpoint (returns "OK").
+- [x] Server Description
+    - [x] Implement `com.atproto.server.describeServer` (returns available user domains).
+
+## 2. Authentication & Account Management (`com.atproto.server`)
+- [x] Account Creation
+    - [x] Implement `com.atproto.server.createAccount`.
+    - [x] Validate handle format (reject invalid characters).
+    - [x] Create DID for new user.
+    - [x] Initialize user repository.
+    - [x] Return access JWT and DID.
+    - [x] MST stuff I think...
+
+- [x] Session Management
+    - [x] Implement `com.atproto.server.createSession` (Login).
+        - [x] Validate identifier (handle/email) and password.
+        - [x] Return access JWT, refresh JWT, and DID.
+    - [x] Implement `com.atproto.server.getSession`.
+        - [x] Verify JWT validity.
+    - [x] Implement `com.atproto.server.refreshSession`.
+    - [x] Implement `com.atproto.server.deleteSession` (Logout).
+        - [x] Invalidate current session/token.
+
+## 3. Repository Operations (`com.atproto.repo`)
+- [ ] Record CRUD
+    - [ ] Implement `com.atproto.repo.createRecord`.
+        - [ ] Generate `rkey` if not provided.
+        - [ ] Validate schema against Lexicon.
+        - [ ] Handle `swapCommit` for optimistic locking.
+    - [ ] Implement `com.atproto.repo.putRecord`.
+        - [ ] Handle create vs update logic.
+        - [ ] Validate `repo` matches authenticated user.
+        - [ ] Validate record schema (e.g., missing required fields).
+    - [ ] Implement `com.atproto.repo.getRecord`.
+        - [ ] Handle missing params (400 Bad Request).
+        - [ ] Handle non-existent record (404 Not Found).
+    - [ ] Implement `com.atproto.repo.deleteRecord`.
+    - [ ] Implement `com.atproto.repo.listRecords`.
+        - [ ] Support pagination (`limit`, `cursor`).
+- [ ] Blob Management
+    - [ ] Implement `com.atproto.repo.uploadBlob`.
+        - [ ] Enforce authentication.
+        - [ ] Validate MIME types (reject unsupported).
+        - [ ] Return blob reference (`$link`).
+- [ ] Repo Meta
+    - [ ] Implement `com.atproto.repo.describeRepo`.
+
+## 4. Actor & Profile (`app.bsky.actor`)
+- [ ] Profile Management
+    - [ ] Implement `app.bsky.actor.getProfile`.
+        - [ ] Resolve handle to DID.
+        - [ ] Return profile record data.
+- [ ] Discovery
+    - [ ] Implement `app.bsky.actor.searchActors`.
+
+## 5. Feed & Timeline (`app.bsky.feed`)
+- [ ] Feed Retrieval
+    - [ ] Implement `app.bsky.feed.getTimeline`.
+    - [ ] Implement `app.bsky.feed.getAuthorFeed`.
+        - [ ] Filter by actor.
+        - [ ] Respect mutes (if viewer is authenticated).
+    - [ ] Implement `app.bsky.feed.getPostThread`.
+        - [ ] Construct thread tree (parents, replies).
+        - [ ] Handle deleted posts (return `notFoundPost` view).
+- [ ] Record Types
+    - [ ] Support `app.bsky.feed.post` record type.
+    - [ ] Support `app.bsky.feed.like` record type.
+    - [ ] Support `app.bsky.embed.images` in posts.
+
+## 6. Social Graph (`app.bsky.graph`)
+- [ ] Relationships
+    - [ ] Implement `app.bsky.graph.getFollows`.
+    - [ ] Implement `app.bsky.graph.getFollowers`.
+    - [ ] Implement `app.bsky.graph.getMutes`.
+    - [ ] Implement `app.bsky.graph.getBlocks`.
+- [ ] Record Types
+    - [ ] Support `app.bsky.graph.follow` record type.
+    - [ ] Support `app.bsky.graph.mute` record type.
+
+## 7. Notifications (`app.bsky.notification`)
+- [ ] Notification Management
+    - [ ] Implement `app.bsky.notification.listNotifications`.
+        - [ ] Aggregate notifications (likes, follows, replies).
+    - [ ] Implement `app.bsky.notification.getUnreadCount`.
+        - [ ] Track read state.
+        - [ ] Reset count on list/read.
+
+## 8. Identity (`com.atproto.identity`)
+- [ ] Resolution
+    - [ ] Implement `com.atproto.identity.resolveHandle`.
+
+## 9. Sync & Federation (`com.atproto.sync`)
+- [ ] Data Export
+    - [ ] Implement `com.atproto.sync.getRepo` (Export CAR file).
+    - [ ] Implement `com.atproto.sync.getBlocks`.
+
+## 10. General Requirements
+- [ ] Validation
+    - [ ] Ensure all endpoints validate input parameters.
+    - [ ] Ensure proper error codes (400, 401, 404, 409).
+- [ ] Concurrency
+    - [ ] Ensure thread safety for repo updates.

docker-compose.yaml

@@ -0,0 +1,50 @@
+services:
+  app:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    image: bspds
+    ports:
+      - "3000:3000"
+    environment:
+      SERVER_HOST: 0.0.0.0
+      SERVER_PORT: 3000
+      DATABASE_URL: postgres://postgres:postgres@db:5432/pds
+      OBJECT_STORAGE_ENDPOINT: http://objsto:9000
+      OBJECT_STORAGE_REGION: us-east-1
+      OBJECT_STORAGE_BUCKET: pds-blobs
+      OBJECT_STORAGE_ACCESS_KEY: minioadmin
+      OBJECT_STORAGE_SECRET_KEY: minioadmin
+      OBJECT_STORAGE_FORCE_PATH_STYLE: "true"
+      JWT_SECRET: your-super-secret-jwt-key-please-change-me
+      PDS_HOSTNAME: localhost:3000
+    depends_on:
+      - db
+      - objsto
+
+  db:
+    image: postgres:latest
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: pds
+    ports:
+      - "5432:5432"
+    volumes:
+      - postgres_data:/var/lib/postgresql
+
+  objsto:
+    image: minio/minio
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+    environment:
+      MINIO_ROOT_USER: minioadmin
+      MINIO_ROOT_PASSWORD: minioadmin
+    volumes:
+      - minio_data:/data
+    command: server /data --console-address ":9001"
+
+volumes:
+  postgres_data:
+  minio_data:

migrations/202512211400_initial_tables.sql

@@ -0,0 +1,80 @@
+-- A very basic schema to get started.
+-- TODO: PRODUCTIONIZE BABY
+
+CREATE TABLE IF NOT EXISTS users (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    handle TEXT NOT NULL UNIQUE,
+    email TEXT NOT NULL UNIQUE,
+    did TEXT NOT NULL UNIQUE,
+    password_hash TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS invite_codes (
+    code TEXT PRIMARY KEY,
+    available_uses INT NOT NULL DEFAULT 1,
+    created_by_user UUID NOT NULL REFERENCES users(id),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS invite_code_uses (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    code TEXT NOT NULL REFERENCES invite_codes(code),
+    used_by_user UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    used_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    UNIQUE(code, used_by_user)
+);
+
+-- OIII THIS TABLE CONTAINS PLAINTEXT PRIVATE KEYS, TODO: encrypt at rest!
+CREATE TABLE IF NOT EXISTS user_keys (
+    user_id UUID PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+    -- Storing as raw bytes
+    -- secp256k1 is 32 bytes
+    key_bytes BYTEA NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS repos (
+    user_id UUID PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+    repo_root_cid TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS blocks (
+    cid BYTEA PRIMARY KEY,
+    data BYTEA NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- A denormalized table to quickly query for records
+-- TODO: Do I actually need this?
+CREATE TABLE IF NOT EXISTS records (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    repo_id UUID NOT NULL REFERENCES repos(user_id) ON DELETE CASCADE,
+    collection TEXT NOT NULL,
+    rkey TEXT NOT NULL,
+    record_cid TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    UNIQUE(repo_id, collection, rkey)
+);
+
+CREATE TABLE IF NOT EXISTS blobs (
+    cid TEXT PRIMARY KEY,
+    mime_type TEXT NOT NULL,
+    size_bytes BIGINT NOT NULL,
+    created_by_user UUID NOT NULL REFERENCES users(id),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+
+    -- The key/path in the S3 bucket
+    storage_key TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS sessions (
+    access_jwt TEXT PRIMARY KEY,
+    refresh_jwt TEXT NOT NULL UNIQUE,
+    did TEXT NOT NULL REFERENCES users(did) ON DELETE CASCADE,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+

src/api/mod.rs

@@ -0,0 +1,2 @@
+pub mod server;
+pub mod repo;

src/api/repo.rs

@@ -0,0 +1,219 @@
+use axum::{
+    extract::State,
+    Json,
+    response::{IntoResponse, Response},
+    http::StatusCode,
+};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use crate::state::AppState;
+use chrono::Utc;
+use sqlx::Row;
+use cid::Cid;
+use std::str::FromStr;
+use jacquard_repo::{mst::Mst, commit::Commit, storage::BlockStore};
+use jacquard::types::{string::{Nsid, Tid}, did::Did, integer::LimitedU32};
+use tracing::error;
+use std::sync::Arc;
+
+#[derive(Deserialize)]
+#[allow(dead_code)]
+pub struct CreateRecordInput {
+    pub repo: String,
+    pub collection: String,
+    pub rkey: Option<String>,
+    pub validate: Option<bool>,
+    pub record: serde_json::Value,
+    #[serde(rename = "swapCommit")]
+    pub swap_commit: Option<String>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateRecordOutput {
+    pub uri: String,
+    pub cid: String,
+}
+
+pub async fn create_record(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+    Json(input): Json<CreateRecordInput>,
+) -> Response {
+    let auth_header = headers.get("Authorization");
+    if auth_header.is_none() {
+        return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationRequired"}))).into_response();
+    }
+    let token = auth_header.unwrap().to_str().unwrap_or("").replace("Bearer ", "");
+
+    if let Err(_) = crate::auth::verify_token(&token) {
+        return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid token"}))).into_response();
+    }
+
+    let session = sqlx::query("SELECT did FROM sessions WHERE access_jwt = $1")
+        .bind(&token)
+        .fetch_optional(&state.db)
+        .await
+        .unwrap_or(None);
+
+    let did = match session {
+        Some(row) => row.get::<String, _>("did"),
+        None => return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed"}))).into_response(),
+    };
+
+    if input.repo != did {
+        return (StatusCode::FORBIDDEN, Json(json!({"error": "InvalidRepo", "message": "Repo does not match authenticated user"}))).into_response();
+    }
+
+    let user_query = sqlx::query("SELECT id FROM users WHERE did = $1")
+        .bind(&did)
+        .fetch_optional(&state.db)
+        .await;
+
+    let user_id: uuid::Uuid = match user_query {
+        Ok(Some(row)) => row.get("id"),
+        _ => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "User not found"}))).into_response(),
+    };
+
+    let repo_root_query = sqlx::query("SELECT repo_root_cid FROM repos WHERE user_id = $1")
+        .bind(user_id)
+        .fetch_optional(&state.db)
+        .await;
+
+    let current_root_cid = match repo_root_query {
+        Ok(Some(row)) => {
+            let cid_str: String = row.get("repo_root_cid");
+            Cid::from_str(&cid_str).ok()
+        },
+        _ => None,
+    };
+
+    if current_root_cid.is_none() {
+         error!("Repo root not found for user {}", did);
+         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Repo root not found"}))).into_response();
+    }
+    let current_root_cid = current_root_cid.unwrap();
+
+    let commit_bytes = match state.block_store.get(&current_root_cid).await {
+        Ok(Some(b)) => b,
+        Ok(None) => {
+             error!("Commit block not found: {}", current_root_cid);
+             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        },
+        Err(e) => {
+             error!("Failed to load commit block: {:?}", e);
+             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let commit = match Commit::from_cbor(&commit_bytes) {
+        Ok(c) => c,
+        Err(e) => {
+             error!("Failed to parse commit: {:?}", e);
+             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let mst_root = commit.data;
+    let store = Arc::new(state.block_store.clone());
+    let mst = Mst::load(store.clone(), mst_root, None);
+
+    let collection_nsid = match input.collection.parse::<Nsid>() {
+        Ok(n) => n,
+        Err(_) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidCollection"}))).into_response(),
+    };
+
+    let rkey = input.rkey.unwrap_or_else(|| {
+        Utc::now().format("%Y%m%d%H%M%S%f").to_string()
+    });
+
+    let mut record_bytes = Vec::new();
+    if let Err(e) = serde_ipld_dagcbor::to_writer(&mut record_bytes, &input.record) {
+        error!("Error serializing record: {:?}", e);
+        return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidRecord", "message": "Failed to serialize record"}))).into_response();
+    }
+
+    let record_cid = match state.block_store.put(&record_bytes).await {
+        Ok(c) => c,
+        Err(e) => {
+             error!("Failed to save record block: {:?}", e);
+             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let key = format!("{}/{}", collection_nsid, rkey);
+    if let Err(e) = mst.update(&key, record_cid).await {
+         error!("Failed to update MST: {:?}", e);
+         return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+    }
+
+    let new_mst_root = match mst.root().await {
+        Ok(c) => c,
+        Err(e) => {
+             error!("Failed to get new MST root: {:?}", e);
+             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let did_obj = match Did::new(&did) {
+        Ok(d) => d,
+        Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Invalid DID"}))).into_response(),
+    };
+
+    let rev = Tid::now(LimitedU32::MIN);
+
+    let new_commit = Commit::new_unsigned(
+        did_obj,
+        new_mst_root,
+        rev,
+        Some(current_root_cid)
+    );
+
+    let new_commit_bytes = match new_commit.to_cbor() {
+        Ok(b) => b,
+        Err(e) => {
+             error!("Failed to serialize new commit: {:?}", e);
+             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let new_root_cid = match state.block_store.put(&new_commit_bytes).await {
+        Ok(c) => c,
+        Err(e) => {
+             error!("Failed to save new commit: {:?}", e);
+             return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let update_repo = sqlx::query("UPDATE repos SET repo_root_cid = $1 WHERE user_id = $2")
+        .bind(new_root_cid.to_string())
+        .bind(user_id)
+        .execute(&state.db)
+        .await;
+
+    if let Err(e) = update_repo {
+        error!("Failed to update repo root in DB: {:?}", e);
+        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+    }
+
+    let record_insert = sqlx::query(
+        "INSERT INTO records (repo_id, collection, rkey, record_cid) VALUES ($1, $2, $3, $4)
+         ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, created_at = NOW()"
+    )
+        .bind(user_id)
+        .bind(&input.collection)
+        .bind(&rkey)
+        .bind(record_cid.to_string())
+        .execute(&state.db)
+        .await;
+
+    if let Err(e) = record_insert {
+        error!("Error inserting record index: {:?}", e);
+    }
+
+    let output = CreateRecordOutput {
+        uri: format!("at://{}/{}/{}", input.repo, input.collection, rkey),
+        cid: record_cid.to_string(),
+    };
+    (StatusCode::OK, Json(output)).into_response()
+}

src/api/server.rs

@@ -0,0 +1,483 @@
+use axum::{
+    extract::State,
+    Json,
+    response::{IntoResponse, Response},
+    http::StatusCode,
+};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use crate::state::AppState;
+use sqlx::Row;
+use bcrypt::{hash, verify, DEFAULT_COST};
+use tracing::{info, error, warn};
+use jacquard_repo::{mst::Mst, commit::Commit, storage::BlockStore};
+use jacquard::types::{string::Tid, did::Did, integer::LimitedU32};
+use std::sync::Arc;
+
+#[derive(Deserialize)]
+pub struct CreateAccountInput {
+    pub handle: String,
+    pub email: String,
+    pub password: String,
+    #[serde(rename = "inviteCode")]
+    pub invite_code: Option<String>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateAccountOutput {
+    pub access_jwt: String,
+    pub refresh_jwt: String,
+    pub handle: String,
+    pub did: String,
+}
+
+pub async fn create_account(
+    State(state): State<AppState>,
+    Json(input): Json<CreateAccountInput>,
+) -> Response {
+    info!("create_account hit: {}", input.handle);
+    if input.handle.contains('!') || input.handle.contains('@') {
+        return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidHandle", "message": "Handle contains invalid characters"}))).into_response();
+    }
+
+    let mut tx = match state.db.begin().await {
+        Ok(tx) => tx,
+        Err(e) => {
+            error!("Error starting transaction: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let exists_query = sqlx::query("SELECT 1 FROM users WHERE handle = $1")
+        .bind(&input.handle)
+        .fetch_optional(&mut *tx)
+        .await;
+
+    match exists_query {
+        Ok(Some(_)) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "HandleTaken", "message": "Handle already taken"}))).into_response(),
+        Err(e) => {
+            error!("Error checking handle: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+        Ok(None) => {}
+    }
+
+    if let Some(code) = &input.invite_code {
+        let invite_query = sqlx::query("SELECT available_uses FROM invite_codes WHERE code = $1 FOR UPDATE")
+            .bind(code)
+            .fetch_optional(&mut *tx)
+            .await;
+
+        match invite_query {
+            Ok(Some(row)) => {
+                let uses: i32 = row.get("available_uses");
+                if uses <= 0 {
+                    return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidInviteCode", "message": "Invite code exhausted"}))).into_response();
+                }
+
+                let update_invite = sqlx::query("UPDATE invite_codes SET available_uses = available_uses - 1 WHERE code = $1")
+                    .bind(code)
+                    .execute(&mut *tx)
+                    .await;
+
+                if let Err(e) = update_invite {
+                    error!("Error updating invite code: {:?}", e);
+                    return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                }
+            },
+            Ok(None) => return (StatusCode::BAD_REQUEST, Json(json!({"error": "InvalidInviteCode", "message": "Invite code not found"}))).into_response(),
+            Err(e) => {
+                error!("Error checking invite code: {:?}", e);
+                return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+            }
+        }
+    }
+
+    let did = format!("did:plc:{}", uuid::Uuid::new_v4());
+
+    let password_hash = match hash(&input.password, DEFAULT_COST) {
+        Ok(h) => h,
+        Err(e) => {
+            error!("Error hashing password: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let user_insert = sqlx::query("INSERT INTO users (handle, email, did, password_hash) VALUES ($1, $2, $3, $4) RETURNING id")
+        .bind(&input.handle)
+        .bind(&input.email)
+        .bind(&did)
+        .bind(&password_hash)
+        .fetch_one(&mut *tx)
+        .await;
+
+    let user_id: uuid::Uuid = match user_insert {
+        Ok(row) => row.get("id"),
+        Err(e) => {
+            error!("Error inserting user: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let store = Arc::new(state.block_store.clone());
+    let mst = Mst::new(store.clone());
+    let mst_root = match mst.root().await {
+        Ok(c) => c,
+        Err(e) => {
+            error!("Error creating MST root: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let did_obj = match Did::new(&did) {
+        Ok(d) => d,
+        Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError", "message": "Invalid DID"}))).into_response(),
+    };
+
+    let rev = Tid::now(LimitedU32::MIN);
+
+    let commit = Commit::new_unsigned(
+        did_obj,
+        mst_root,
+        rev,
+        None
+    );
+
+    let commit_bytes = match commit.to_cbor() {
+        Ok(b) => b,
+        Err(e) => {
+            error!("Error serializing genesis commit: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let commit_cid = match state.block_store.put(&commit_bytes).await {
+        Ok(c) => c,
+        Err(e) => {
+            error!("Error saving genesis commit: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    };
+
+    let repo_insert = sqlx::query("INSERT INTO repos (user_id, repo_root_cid) VALUES ($1, $2)")
+        .bind(user_id)
+        .bind(commit_cid.to_string())
+        .execute(&mut *tx)
+        .await;
+
+    if let Err(e) = repo_insert {
+        error!("Error initializing repo: {:?}", e);
+        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+    }
+
+    if let Some(code) = &input.invite_code {
+        let use_insert = sqlx::query("INSERT INTO invite_code_uses (code, used_by_user) VALUES ($1, $2)")
+            .bind(code)
+            .bind(user_id)
+            .execute(&mut *tx)
+            .await;
+
+        if let Err(e) = use_insert {
+            error!("Error recording invite usage: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    }
+
+    let access_jwt = crate::auth::create_access_token(&did).map_err(|e| {
+        error!("Error creating access token: {:?}", e);
+        (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response()
+    });
+    let access_jwt = match access_jwt {
+        Ok(t) => t,
+        Err(r) => return r,
+    };
+
+    let refresh_jwt = crate::auth::create_refresh_token(&did).map_err(|e| {
+        error!("Error creating refresh token: {:?}", e);
+        (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response()
+    });
+    let refresh_jwt = match refresh_jwt {
+        Ok(t) => t,
+        Err(r) => return r,
+    };
+
+    let session_insert = sqlx::query("INSERT INTO sessions (access_jwt, refresh_jwt, did) VALUES ($1, $2, $3)")
+        .bind(&access_jwt)
+        .bind(&refresh_jwt)
+        .bind(&did)
+        .execute(&mut *tx)
+        .await;
+
+    if let Err(e) = session_insert {
+        error!("Error inserting session: {:?}", e);
+        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+    }
+
+    if let Err(e) = tx.commit().await {
+        error!("Error committing transaction: {:?}", e);
+        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+    }
+
+    (StatusCode::OK, Json(CreateAccountOutput {
+        access_jwt,
+        refresh_jwt,
+        handle: input.handle,
+        did,
+    })).into_response()
+}
+
+#[derive(Deserialize)]
+pub struct CreateSessionInput {
+    pub identifier: String,
+    pub password: String,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateSessionOutput {
+    pub access_jwt: String,
+    pub refresh_jwt: String,
+    pub handle: String,
+    pub did: String,
+}
+
+pub async fn create_session(
+    State(state): State<AppState>,
+    Json(input): Json<CreateSessionInput>,
+) -> Response {
+    info!("create_session: identifier='{}'", input.identifier);
+
+    let user_row = sqlx::query("SELECT did, handle, password_hash FROM users WHERE handle = $1 OR email = $1")
+        .bind(&input.identifier)
+        .fetch_optional(&state.db)
+        .await;
+
+    match user_row {
+        Ok(Some(row)) => {
+            let stored_hash: String = row.get("password_hash");
+
+            if verify(&input.password, &stored_hash).unwrap_or(false) {
+                let did: String = row.get("did");
+                let handle: String = row.get("handle");
+
+                let access_jwt = match crate::auth::create_access_token(&did) {
+                    Ok(t) => t,
+                    Err(e) => {
+                        error!("Failed to create access token: {:?}", e);
+                        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                    }
+                };
+
+                let refresh_jwt = match crate::auth::create_refresh_token(&did) {
+                    Ok(t) => t,
+                    Err(e) => {
+                        error!("Failed to create refresh token: {:?}", e);
+                        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                    }
+                };
+
+                let session_insert = sqlx::query("INSERT INTO sessions (access_jwt, refresh_jwt, did) VALUES ($1, $2, $3)")
+                    .bind(&access_jwt)
+                    .bind(&refresh_jwt)
+                    .bind(&did)
+                    .execute(&state.db)
+                    .await;
+
+                match session_insert {
+                    Ok(_) => {
+                        return (StatusCode::OK, Json(CreateSessionOutput {
+                            access_jwt,
+                            refresh_jwt,
+                            handle,
+                            did,
+                        })).into_response();
+                    },
+                    Err(e) => {
+                        error!("Failed to insert session: {:?}", e);
+                        return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                    }
+                }
+            } else {
+                warn!("Password verification failed for identifier: {}", input.identifier);
+            }
+        },
+        Ok(None) => {
+            warn!("User not found for identifier: {}", input.identifier);
+        },
+        Err(e) => {
+            error!("Database error fetching user: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    }
+
+    (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid identifier or password"}))).into_response()
+}
+
+pub async fn get_session(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+) -> Response {
+    let auth_header = headers.get("Authorization");
+    if auth_header.is_none() {
+        return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationRequired"}))).into_response();
+    }
+
+    let token = auth_header.unwrap().to_str().unwrap_or("").replace("Bearer ", "");
+
+    if let Err(_) = crate::auth::verify_token(&token) {
+        return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid token"}))).into_response();
+    }
+
+    let result = sqlx::query(
+        r#"
+        SELECT u.handle, u.did, u.email
+        FROM sessions s
+        JOIN users u ON s.did = u.did
+        WHERE s.access_jwt = $1
+        "#
+    )
+    .bind(token)
+    .fetch_optional(&state.db)
+    .await;
+
+    match result {
+        Ok(Some(row)) => {
+            let handle: String = row.get("handle");
+            let did: String = row.get("did");
+            let email: String = row.get("email");
+
+            return (StatusCode::OK, Json(json!({
+                "handle": handle,
+                "did": did,
+                "email": email,
+                "didDoc": {}
+            }))).into_response();
+        },
+        Ok(None) => {
+            return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed"}))).into_response();
+        },
+        Err(e) => {
+            error!("Database error in get_session: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    }
+}
+
+pub async fn delete_session(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+) -> Response {
+    let auth_header = headers.get("Authorization");
+    if auth_header.is_none() {
+        return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationRequired"}))).into_response();
+    }
+
+    let token = auth_header.unwrap().to_str().unwrap_or("").replace("Bearer ", "");
+
+    let result = sqlx::query("DELETE FROM sessions WHERE access_jwt = $1")
+        .bind(token)
+        .execute(&state.db)
+        .await;
+
+    match result {
+        Ok(res) => {
+            if res.rows_affected() > 0 {
+                return (StatusCode::OK, Json(json!({}))).into_response();
+            }
+        },
+        Err(e) => {
+            error!("Database error in delete_session: {:?}", e);
+        }
+    }
+
+    (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed"}))).into_response()
+}
+
+pub async fn refresh_session(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+) -> Response {
+    let auth_header = headers.get("Authorization");
+    if auth_header.is_none() {
+        return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationRequired"}))).into_response();
+    }
+
+    let refresh_token = auth_header.unwrap().to_str().unwrap_or("").replace("Bearer ", "");
+
+    if let Err(_) = crate::auth::verify_token(&refresh_token) {
+        return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid refresh token"}))).into_response();
+    }
+
+    let session = sqlx::query("SELECT did FROM sessions WHERE refresh_jwt = $1")
+        .bind(&refresh_token)
+        .fetch_optional(&state.db)
+        .await;
+
+    match session {
+        Ok(Some(session_row)) => {
+            let did: String = session_row.get("did");
+            let new_access_jwt = match crate::auth::create_access_token(&did) {
+                Ok(t) => t,
+                Err(e) => {
+                    error!("Failed to create access token: {:?}", e);
+                    return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                }
+            };
+            let new_refresh_jwt = match crate::auth::create_refresh_token(&did) {
+                Ok(t) => t,
+                Err(e) => {
+                    error!("Failed to create refresh token: {:?}", e);
+                    return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                }
+            };
+
+            let update = sqlx::query("UPDATE sessions SET access_jwt = $1, refresh_jwt = $2 WHERE refresh_jwt = $3")
+                .bind(&new_access_jwt)
+                .bind(&new_refresh_jwt)
+                .bind(&refresh_token)
+                .execute(&state.db)
+                .await;
+
+            match update {
+                Ok(_) => {
+                     let user = sqlx::query("SELECT handle FROM users WHERE did = $1")
+                        .bind(&did)
+                        .fetch_optional(&state.db)
+                        .await;
+
+                    match user {
+                        Ok(Some(u)) => {
+                            let handle: String = u.get("handle");
+                            return (StatusCode::OK, Json(json!({
+                                "accessJwt": new_access_jwt,
+                                "refreshJwt": new_refresh_jwt,
+                                "handle": handle,
+                                "did": did
+                            }))).into_response();
+                        },
+                        Ok(None) => {
+                            error!("User not found for existing session: {}", did);
+                            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                        },
+                        Err(e) => {
+                            error!("Database error fetching user: {:?}", e);
+                            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                        }
+                    }
+                },
+                Err(e) => {
+                    error!("Database error updating session: {:?}", e);
+                    return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+                }
+            }
+        },
+        Ok(None) => {
+            return (StatusCode::UNAUTHORIZED, Json(json!({"error": "AuthenticationFailed", "message": "Invalid refresh token"}))).into_response();
+        },
+        Err(e) => {
+            error!("Database error fetching session: {:?}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "InternalError"}))).into_response();
+        }
+    }
+}

src/auth.rs

@@ -0,0 +1,59 @@
+use jsonwebtoken::{encode, decode, Header, Validation, EncodingKey, DecodingKey, TokenData};
+use serde::{Deserialize, Serialize};
+use chrono::{Utc, Duration};
+use std::env;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Claims {
+    // DID type shit
+    pub sub: String,
+    pub exp: usize,
+    pub iat: usize,
+    pub scope: String,
+    pub jti: String,
+}
+
+pub fn create_access_token(did: &str) -> Result<String, jsonwebtoken::errors::Error> {
+    let secret = env::var("JWT_SECRET").unwrap_or_else(|_| "secret".to_string());
+    let expiration = Utc::now()
+        .checked_add_signed(Duration::minutes(15))
+        .expect("valid timestamp")
+        .timestamp();
+
+    let claims = Claims {
+        sub: did.to_owned(),
+        exp: expiration as usize,
+        iat: Utc::now().timestamp() as usize,
+        scope: "access".to_string(),
+        jti: uuid::Uuid::new_v4().to_string(),
+    };
+
+    encode(&Header::default(), &claims, &EncodingKey::from_secret(secret.as_ref()))
+}
+
+pub fn create_refresh_token(did: &str) -> Result<String, jsonwebtoken::errors::Error> {
+    let secret = env::var("JWT_SECRET").unwrap_or_else(|_| "secret".to_string());
+    let expiration = Utc::now()
+        .checked_add_signed(Duration::days(7))
+        .expect("valid timestamp")
+        .timestamp();
+
+    let claims = Claims {
+        sub: did.to_owned(),
+        exp: expiration as usize,
+        iat: Utc::now().timestamp() as usize,
+        scope: "refresh".to_string(),
+        jti: uuid::Uuid::new_v4().to_string(),
+    };
+
+    encode(&Header::default(), &claims, &EncodingKey::from_secret(secret.as_ref()))
+}
+
+pub fn verify_token(token: &str) -> Result<TokenData<Claims>, jsonwebtoken::errors::Error> {
+    let secret = env::var("JWT_SECRET").unwrap_or_else(|_| "secret".to_string());
+    decode::<Claims>(
+        token,
+        &DecodingKey::from_secret(secret.as_ref()),
+        &Validation::default(),
+    )
+}

src/main.rs

@@ -0,0 +1,73 @@
+mod api;
+mod state;
+mod auth;
+mod repo;
+
+use axum::{
+    extract::State,
+    routing::{get, post},
+    Router,
+    Json,
+    response::IntoResponse,
+    http::StatusCode,
+};
+use serde_json::json;
+use std::net::SocketAddr;
+use state::AppState;
+use tracing::{info, error};
+
+#[tokio::main]
+async fn main() {
+    dotenvy::dotenv().ok();
+    tracing_subscriber::fmt::init();
+
+    let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
+
+    let pool = sqlx::postgres::PgPoolOptions::new()
+        .max_connections(5)
+        .connect(&database_url)
+        .await
+        .expect("Failed to connect to Postgres");
+
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("Failed to run migrations");
+
+    let state = AppState::new(pool);
+
+    let app = Router::new()
+        .route("/health", get(health))
+        .route("/xrpc/com.atproto.server.describeServer", get(describe_server))
+        .route("/xrpc/com.atproto.server.createAccount", post(api::server::create_account))
+        .route("/xrpc/com.atproto.server.createSession", post(api::server::create_session))
+        .route("/xrpc/com.atproto.server.getSession", get(api::server::get_session))
+        .route("/xrpc/com.atproto.server.deleteSession", post(api::server::delete_session))
+        .route("/xrpc/com.atproto.server.refreshSession", post(api::server::refresh_session))
+        .route("/xrpc/com.atproto.repo.createRecord", post(api::repo::create_record))
+        .with_state(state);
+
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
+    info!("listening on {}", addr);
+    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
+    axum::serve(listener, app).await.unwrap();
+}
+
+async fn health(State(state): State<AppState>) -> impl IntoResponse {
+    match sqlx::query("SELECT 1").execute(&state.db).await {
+        Ok(_) => (StatusCode::OK, "OK"),
+        Err(e) => {
+            error!("Health check failed: {:?}", e);
+            (StatusCode::SERVICE_UNAVAILABLE, "Service Unavailable")
+        }
+    }
+}
+
+async fn describe_server() -> impl IntoResponse {
+    let domains_str = std::env::var("AVAILABLE_USER_DOMAINS").unwrap_or_else(|_| "example.com".to_string());
+    let domains: Vec<&str> = domains_str.split(',').map(|s| s.trim()).collect();
+
+    Json(json!({
+        "availableUserDomains": domains
+    }))
+}

src/repo/mod.rs

@@ -0,0 +1,94 @@
+use jacquard_repo::storage::BlockStore;
+use jacquard_repo::error::RepoError;
+use jacquard_repo::repo::CommitData;
+use cid::Cid;
+use sqlx::{PgPool, Row};
+use bytes::Bytes;
+use sha2::{Sha256, Digest};
+use multihash::Multihash;
+
+#[derive(Clone)]
+pub struct PostgresBlockStore {
+    pool: PgPool,
+}
+
+impl PostgresBlockStore {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+impl BlockStore for PostgresBlockStore {
+    async fn get(&self, cid: &Cid) -> Result<Option<Bytes>, RepoError> {
+        let cid_bytes = cid.to_bytes();
+        let row = sqlx::query("SELECT data FROM blocks WHERE cid = $1")
+            .bind(cid_bytes)
+            .fetch_optional(&self.pool)
+            .await
+            .map_err(|e| RepoError::storage(e))?;
+
+        match row {
+            Some(row) => {
+                let data: Vec<u8> = row.get("data");
+                Ok(Some(Bytes::from(data)))
+            },
+            None => Ok(None),
+        }
+    }
+
+    async fn put(&self, data: &[u8]) -> Result<Cid, RepoError> {
+        let mut hasher = Sha256::new();
+        hasher.update(data);
+        let hash = hasher.finalize();
+        let multihash = Multihash::wrap(0x12, &hash).unwrap();
+        let cid = Cid::new_v1(0x71, multihash);
+        let cid_bytes = cid.to_bytes();
+
+        sqlx::query("INSERT INTO blocks (cid, data) VALUES ($1, $2) ON CONFLICT (cid) DO NOTHING")
+            .bind(cid_bytes)
+            .bind(data)
+            .execute(&self.pool)
+            .await
+            .map_err(|e| RepoError::storage(e))?;
+
+        Ok(cid)
+    }
+
+    async fn has(&self, cid: &Cid) -> Result<bool, RepoError> {
+        let cid_bytes = cid.to_bytes();
+        let row = sqlx::query("SELECT 1 FROM blocks WHERE cid = $1")
+            .bind(cid_bytes)
+            .fetch_optional(&self.pool)
+            .await
+            .map_err(|e| RepoError::storage(e))?;
+
+        Ok(row.is_some())
+    }
+
+    async fn put_many(&self, blocks: impl IntoIterator<Item = (Cid, Bytes)> + Send) -> Result<(), RepoError> {
+        let blocks: Vec<_> = blocks.into_iter().collect();
+        for (cid, data) in blocks {
+            let cid_bytes = cid.to_bytes();
+            sqlx::query("INSERT INTO blocks (cid, data) VALUES ($1, $2) ON CONFLICT (cid) DO NOTHING")
+                .bind(cid_bytes)
+                .bind(data.as_ref())
+                .execute(&self.pool)
+                .await
+                .map_err(|e| RepoError::storage(e))?;
+        }
+        Ok(())
+    }
+
+    async fn get_many(&self, cids: &[Cid]) -> Result<Vec<Option<Bytes>>, RepoError> {
+        let mut results = Vec::new();
+        for cid in cids {
+            results.push(self.get(cid).await?);
+        }
+        Ok(results)
+    }
+
+    async fn apply_commit(&self, commit: CommitData) -> Result<(), RepoError> {
+        self.put_many(commit.blocks).await?;
+        Ok(())
+    }
+}

src/state.rs

@@ -0,0 +1,15 @@
+use sqlx::PgPool;
+use crate::repo::PostgresBlockStore;
+
+#[derive(Clone)]
+pub struct AppState {
+    pub db: PgPool,
+    pub block_store: PostgresBlockStore,
+}
+
+impl AppState {
+    pub fn new(db: PgPool) -> Self {
+        let block_store = PostgresBlockStore::new(db.clone());
+        Self { db, block_store }
+    }
+}

tests/actor.rs

@@ -0,0 +1,36 @@
+mod common;
+use common::*;
+use reqwest::StatusCode;
+
+#[tokio::test]
+async fn test_get_profile() {
+    let client = client();
+    let params = [
+        ("actor", AUTH_DID),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.actor.getProfile", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_search_actors() {
+    let client = client();
+    let params = [
+        ("q", "test"),
+        ("limit", "10"),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.actor.searchActors", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}

tests/common/mod.rs

@@ -0,0 +1,101 @@
+use reqwest::{header, Client, StatusCode};
+use serde_json::{json, Value};
+use chrono::Utc;
+#[allow(unused_imports)]
+use std::collections::HashMap;
+#[allow(unused_imports)]
+use std::time::Duration;
+
+pub const BASE_URL: &str = "http://127.0.0.1:3000";
+#[allow(dead_code)]
+pub const AUTH_TOKEN: &str = "test-token";
+#[allow(dead_code)]
+pub const BAD_AUTH_TOKEN: &str = "bad-token";
+#[allow(dead_code)]
+pub const AUTH_DID: &str = "did:plc:fake";
+#[allow(dead_code)]
+pub const TARGET_DID: &str = "did:plc:target";
+
+pub fn client() -> Client {
+    Client::new()
+}
+
+#[allow(dead_code)]
+pub async fn upload_test_blob(client: &Client, data: &'static str, mime: &'static str) -> Value {
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", BASE_URL))
+        .header(header::CONTENT_TYPE, mime)
+        .bearer_auth(AUTH_TOKEN)
+        .body(data)
+        .send()
+        .await
+        .expect("Failed to send uploadBlob request");
+
+    assert_eq!(res.status(), StatusCode::OK, "Failed to upload blob");
+    let body: Value = res.json().await.expect("Blob upload response was not JSON");
+    body["blob"].clone()
+}
+
+
+#[allow(dead_code)]
+pub async fn create_test_post(
+    client: &Client,
+    text: &str,
+    reply_to: Option<Value>
+) -> (String, String, String) {
+    let collection = "app.bsky.feed.post";
+    let mut record = json!({
+        "$type": collection,
+        "text": text,
+        "createdAt": Utc::now().to_rfc3339()
+    });
+
+    if let Some(reply_obj) = reply_to {
+        record["reply"] = reply_obj;
+    }
+
+    let payload = json!({
+        "repo": AUTH_DID,
+        "collection": collection,
+        "record": record
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.createRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send createRecord");
+
+    assert_eq!(res.status(), StatusCode::OK, "Failed to create post record");
+    let body: Value = res.json().await.expect("createRecord response was not JSON");
+
+    let uri = body["uri"].as_str().expect("Response had no URI").to_string();
+    let cid = body["cid"].as_str().expect("Response had no CID").to_string();
+    let rkey = uri.split('/').last().expect("URI was malformed").to_string();
+
+    (uri, cid, rkey)
+}
+
+pub async fn create_account_and_login(client: &Client) -> (String, String) {
+    let handle = format!("user_{}", uuid::Uuid::new_v4());
+    let payload = json!({
+        "handle": handle,
+        "email": format!("{}@example.com", handle),
+        "password": "password"
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.server.createAccount", BASE_URL))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to create account");
+
+    if res.status() != StatusCode::OK {
+        panic!("Failed to create account: {:?}", res.text().await);
+    }
+
+    let body: Value = res.json().await.expect("Invalid JSON");
+    let access_jwt = body["accessJwt"].as_str().expect("No accessJwt").to_string();
+    let did = body["did"].as_str().expect("No did").to_string();
+    (access_jwt, did)
+}

tests/feed.rs

@@ -0,0 +1,53 @@
+mod common;
+use common::*;
+use reqwest::StatusCode;
+
+use std::collections::HashMap;
+
+#[tokio::test]
+async fn test_get_timeline() {
+    let client = client();
+    let params = [("limit", "30")];
+    let res = client.get(format!("{}/xrpc/app.bsky.feed.getTimeline", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_get_author_feed() {
+    let client = client();
+    let params = [
+        ("actor", AUTH_DID),
+        ("limit", "30")
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.feed.getAuthorFeed", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_get_post_thread() {
+    let client = client();
+    let mut params = HashMap::new();
+    params.insert("uri", "at://did:plc:other/app.bsky.feed.post/3k12345");
+    params.insert("depth", "5");
+
+    let res = client.get(format!("{}/xrpc/app.bsky.feed.getPostThread", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}

tests/graph.rs

@@ -0,0 +1,68 @@
+mod common;
+use common::*;
+use reqwest::StatusCode;
+
+#[tokio::test]
+async fn test_get_follows() {
+    let client = client();
+    let params = [
+        ("actor", AUTH_DID),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.graph.getFollows", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_get_followers() {
+    let client = client();
+    let params = [
+        ("actor", AUTH_DID),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.graph.getFollowers", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_get_mutes() {
+    let client = client();
+    let params = [
+        ("limit", "25"),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.graph.getMutes", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+// User blocks, ie. not repo blocks ya know
+async fn test_get_user_blocks() {
+    let client = client();
+    let params = [
+        ("limit", "25"),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.graph.getBlocks", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}

tests/identity.rs

@@ -0,0 +1,18 @@
+mod common;
+use common::*;
+use reqwest::StatusCode;
+
+#[tokio::test]
+async fn test_resolve_handle() {
+    let client = client();
+    let params = [
+        ("handle", "bsky.app"),
+    ];
+    let res = client.get(format!("{}/xrpc/com.atproto.identity.resolveHandle", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}

tests/lifecycle.rs

@@ -0,0 +1,936 @@
+mod common;
+use common::*;
+
+use reqwest::StatusCode;
+use serde_json::{json, Value};
+use chrono::Utc;
+use std::time::Duration;
+
+use reqwest::Client;
+#[allow(unused_imports)]
+use std::collections::HashMap;
+
+#[tokio::test]
+async fn test_post_crud_lifecycle() {
+    let client = client();
+    let collection = "app.bsky.feed.post";
+
+    let rkey = format!("e2e_lifecycle_{}", Utc::now().timestamp_millis());
+    let now = Utc::now().to_rfc3339();
+
+    let original_text = "Hello from the lifecycle test!";
+    let create_payload = json!({
+        "repo": AUTH_DID,
+        "collection": collection,
+        "rkey": rkey,
+        "record": {
+            "$type": collection,
+            "text": original_text,
+            "createdAt": now
+        }
+    });
+
+    let create_res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&create_payload)
+        .send()
+        .await
+        .expect("Failed to send create request");
+
+    assert_eq!(create_res.status(), StatusCode::OK, "Failed to create record");
+    let create_body: Value = create_res.json().await.expect("create response was not JSON");
+    let uri = create_body["uri"].as_str().unwrap();
+
+
+    let params = [
+        ("repo", AUTH_DID),
+        ("collection", collection),
+        ("rkey", &rkey),
+    ];
+    let get_res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send get request");
+
+    assert_eq!(get_res.status(), StatusCode::OK, "Failed to get record after create");
+    let get_body: Value = get_res.json().await.expect("get response was not JSON");
+    assert_eq!(get_body["uri"], uri);
+    assert_eq!(get_body["value"]["text"], original_text);
+
+
+    let updated_text = "This post has been updated.";
+    let update_payload = json!({
+        "repo": AUTH_DID,
+        "collection": collection,
+        "rkey": rkey,
+        "record": {
+            "$type": collection,
+            "text": updated_text,
+            "createdAt": now
+        }
+    });
+
+    let update_res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&update_payload)
+        .send()
+        .await
+        .expect("Failed to send update request");
+
+    assert_eq!(update_res.status(), StatusCode::OK, "Failed to update record");
+
+
+    let get_updated_res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send get-after-update request");
+
+    assert_eq!(get_updated_res.status(), StatusCode::OK, "Failed to get record after update");
+    let get_updated_body: Value = get_updated_res.json().await.expect("get-updated response was not JSON");
+    assert_eq!(get_updated_body["value"]["text"], updated_text, "Text was not updated");
+
+
+    let delete_payload = json!({
+        "repo": AUTH_DID,
+        "collection": collection,
+        "rkey": rkey
+    });
+
+    let delete_res = client.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&delete_payload)
+        .send()
+        .await
+        .expect("Failed to send delete request");
+
+    assert_eq!(delete_res.status(), StatusCode::OK, "Failed to delete record");
+
+
+    let get_deleted_res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send get-after-delete request");
+
+    assert_eq!(get_deleted_res.status(), StatusCode::NOT_FOUND, "Record was found, but it should be deleted");
+}
+
+#[tokio::test]
+async fn test_post_with_image_lifecycle() {
+    let client = client();
+
+    let now_str = Utc::now().to_rfc3339();
+    let fake_image_data = format!("This is a fake PNG for test at {}", now_str);
+
+    let image_blob = upload_test_blob(
+        &client,
+        Box::leak(fake_image_data.into_boxed_str()),
+        "image/png"
+    ).await;
+
+    let blob_ref = image_blob["ref"].clone();
+    assert!(blob_ref.is_object(), "Blob ref is not an object");
+
+
+    let collection = "app.bsky.feed.post";
+    let rkey = format!("e2e_image_post_{}", Utc::now().timestamp_millis());
+
+    let create_payload = json!({
+        "repo": AUTH_DID,
+        "collection": collection,
+        "rkey": rkey,
+        "record": {
+            "$type": collection,
+            "text": "Check out this image!",
+            "createdAt": Utc::now().to_rfc3339(),
+            "embed": {
+                "$type": "app.bsky.embed.images",
+                "images": [
+                    {
+                        "image": image_blob,
+                        "alt": "A test image"
+                    }
+                ]
+            }
+        }
+    });
+
+    let create_res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&create_payload)
+        .send()
+        .await
+        .expect("Failed to create image post");
+
+    assert_eq!(create_res.status(), StatusCode::OK, "Failed to create post with image");
+
+
+    let params = [
+        ("repo", AUTH_DID),
+        ("collection", collection),
+        ("rkey", &rkey),
+    ];
+    let get_res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to get image post");
+
+    assert_eq!(get_res.status(), StatusCode::OK, "Failed to get image post");
+    let get_body: Value = get_res.json().await.expect("get image post was not JSON");
+
+    let embed_image = &get_body["value"]["embed"]["images"][0]["image"];
+    assert!(embed_image.is_object(), "Embedded image is missing");
+    assert_eq!(embed_image["ref"], blob_ref, "Embedded blob ref does not match uploaded ref");
+}
+
+#[tokio::test]
+async fn test_graph_lifecycle_follow_unfollow() {
+    let client = client();
+    let collection = "app.bsky.graph.follow";
+
+    let create_payload = json!({
+        "repo": AUTH_DID,
+        "collection": collection,
+        // "rkey" is omitted, server will generate it right?
+        "record": {
+            "$type": collection,
+            "subject": TARGET_DID,
+            "createdAt": Utc::now().to_rfc3339()
+        }
+    });
+
+    let create_res = client.post(format!("{}/xrpc/com.atproto.repo.createRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&create_payload)
+        .send()
+        .await
+        .expect("Failed to send follow createRecord");
+
+    assert_eq!(create_res.status(), StatusCode::OK, "Failed to create follow record");
+    let create_body: Value = create_res.json().await.expect("create follow response was not JSON");
+    let follow_uri = create_body["uri"].as_str().expect("Response had no URI");
+
+    let rkey = follow_uri.split('/').last().expect("URI was malformed");
+
+
+    let params_get_follows = [
+        ("actor", AUTH_DID),
+    ];
+    let get_follows_res = client.get(format!("{}/xrpc/app.bsky.graph.getFollows", BASE_URL))
+        .query(&params_get_follows)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send getFollows");
+
+    assert_eq!(get_follows_res.status(), StatusCode::OK, "getFollows did not return 200");
+    let get_follows_body: Value = get_follows_res.json().await.expect("getFollows response was not JSON");
+
+    let follows_list = get_follows_body["follows"].as_array().expect("follows key was not an array");
+    let is_following = follows_list.iter().any(|actor| {
+        actor["did"].as_str() == Some(TARGET_DID)
+    });
+
+    assert!(is_following, "getFollows list did not contain the target DID");
+
+
+    let delete_payload = json!({
+        "repo": AUTH_DID,
+        "collection": collection,
+        "rkey": rkey
+    });
+
+    let delete_res = client.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&delete_payload)
+        .send()
+        .await
+        .expect("Failed to send unfollow deleteRecord");
+
+    assert_eq!(delete_res.status(), StatusCode::OK, "Failed to delete follow record");
+
+
+    let get_unfollowed_res = client.get(format!("{}/xrpc/app.bsky.graph.getFollows", BASE_URL))
+        .query(&params_get_follows)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send getFollows after delete");
+
+    assert_eq!(get_unfollowed_res.status(), StatusCode::OK, "getFollows (after delete) did not return 200");
+    let get_unfollowed_body: Value = get_unfollowed_res.json().await.expect("getFollows (after delete) was not JSON");
+
+    let follows_list_after = get_unfollowed_body["follows"].as_array().expect("follows key was not an array");
+    let is_still_following = follows_list_after.iter().any(|actor| {
+        actor["did"].as_str() == Some(TARGET_DID)
+    });
+
+    assert!(!is_still_following, "getFollows list *still* contains the target DID after unfollow");
+}
+
+#[tokio::test]
+async fn test_list_records_pagination() {
+    let client = client();
+    let collection = "app.bsky.feed.post";
+    let mut created_rkeys = Vec::new();
+
+    for i in 0..3 {
+        let rkey = format!("e2e_pagination_{}", Utc::now().timestamp_millis());
+        let payload = json!({
+            "repo": AUTH_DID,
+            "collection": collection,
+            "rkey": rkey,
+            "record": {
+                "$type": collection,
+                "text": format!("Pagination test post #{}", i),
+                "createdAt": Utc::now().to_rfc3339()
+            }
+        });
+
+        let res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+            .bearer_auth(AUTH_TOKEN)
+            .json(&payload)
+            .send()
+            .await
+            .expect("Failed to create pagination post");
+
+        assert_eq!(res.status(), StatusCode::OK, "Failed to create post for pagination test");
+        created_rkeys.push(rkey);
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+
+    let params_page1 = [
+        ("repo", AUTH_DID),
+        ("collection", collection),
+        ("limit", "2"),
+    ];
+
+    let page1_res = client.get(format!("{}/xrpc/com.atproto.repo.listRecords", BASE_URL))
+        .query(&params_page1)
+        .send()
+        .await
+        .expect("Failed to send listRecords (page 1)");
+
+    assert_eq!(page1_res.status(), StatusCode::OK, "listRecords (page 1) failed");
+    let page1_body: Value = page1_res.json().await.expect("listRecords (page 1) was not JSON");
+
+    let page1_records = page1_body["records"].as_array().expect("records was not an array");
+    assert_eq!(page1_records.len(), 2, "Page 1 did not return 2 records");
+
+    let cursor = page1_body["cursor"].as_str().expect("Page 1 did not have a cursor");
+
+
+    let params_page2 = [
+        ("repo", AUTH_DID),
+        ("collection", collection),
+        ("limit", "2"),
+        ("cursor", cursor),
+    ];
+
+    let page2_res = client.get(format!("{}/xrpc/com.atproto.repo.listRecords", BASE_URL))
+        .query(&params_page2)
+        .send()
+        .await
+        .expect("Failed to send listRecords (page 2)");
+
+    assert_eq!(page2_res.status(), StatusCode::OK, "listRecords (page 2) failed");
+    let page2_body: Value = page2_res.json().await.expect("listRecords (page 2) was not JSON");
+
+    let page2_records = page2_body["records"].as_array().expect("records was not an array");
+    assert_eq!(page2_records.len(), 1, "Page 2 did not return 1 record");
+
+    assert!(page2_body["cursor"].is_null() || page2_body["cursor"].as_str().is_none(), "Page 2 should not have a cursor");
+
+
+    for rkey in created_rkeys {
+        let delete_payload = json!({
+            "repo": AUTH_DID,
+            "collection": collection,
+            "rkey": rkey
+        });
+        client.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", BASE_URL))
+            .bearer_auth(AUTH_TOKEN)
+            .json(&delete_payload)
+            .send()
+            .await
+            .expect("Failed to cleanup pagination post");
+    }
+}
+
+#[tokio::test]
+async fn test_reply_thread_lifecycle() {
+    let client = client();
+
+    let (root_uri, root_cid, root_rkey) = create_test_post(
+        &client,
+        "This is the root of the thread",
+        None
+    ).await;
+
+
+    let reply_ref = json!({
+        "root": { "uri": root_uri.clone(), "cid": root_cid.clone() },
+        "parent": { "uri": root_uri.clone(), "cid": root_cid.clone() }
+    });
+
+    let (reply_uri, _reply_cid, reply_rkey) = create_test_post(
+        &client,
+        "This is a reply!",
+        Some(reply_ref)
+    ).await;
+
+
+    let params = [
+        ("uri", &root_uri),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.feed.getPostThread", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send getPostThread");
+
+    assert_eq!(res.status(), StatusCode::OK, "getPostThread did not return 200");
+    let body: Value = res.json().await.expect("getPostThread response was not JSON");
+
+    assert_eq!(body["thread"]["$type"], "app.bsky.feed.defs#threadViewPost");
+    assert_eq!(body["thread"]["post"]["uri"], root_uri);
+
+    let replies = body["thread"]["replies"].as_array().expect("replies was not an array");
+    assert!(!replies.is_empty(), "Replies array is empty, but should contain the reply");
+
+    let found_reply = replies.iter().find(|r| {
+        r["post"]["uri"] == reply_uri
+    });
+
+    assert!(found_reply.is_some(), "Our specific reply was not found in the thread's replies");
+
+
+    let collection = "app.bsky.feed.post";
+    client.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&json!({ "repo": AUTH_DID, "collection": collection, "rkey": reply_rkey }))
+        .send().await.expect("Failed to delete reply");
+
+    client.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .json(&json!({ "repo": AUTH_DID, "collection": collection, "rkey": root_rkey }))
+        .send().await.expect("Failed to delete root post");
+}
+
+#[tokio::test]
+async fn test_account_journey_lifecycle() {
+    let client = client();
+
+    let ts = Utc::now().timestamp_millis();
+    let handle = format!("e2e-user-{}.test", ts);
+    let email = format!("e2e-user-{}@test.com", ts);
+    let password = "e2e-password-123";
+
+    let create_account_payload = json!({
+        "handle": handle,
+        "email": email,
+        "password": password
+    });
+
+    let create_res = client.post(format!("{}/xrpc/com.atproto.server.createAccount", BASE_URL))
+        .json(&create_account_payload)
+        .send()
+        .await
+        .expect("Failed to send createAccount");
+
+    assert_eq!(create_res.status(), StatusCode::OK, "Failed to create account");
+    let create_body: Value = create_res.json().await.expect("createAccount response was not JSON");
+
+    let new_did = create_body["did"].as_str().expect("Response had no DID").to_string();
+    let _new_jwt = create_body["accessJwt"].as_str().expect("Response had no accessJwt").to_string();
+    assert_eq!(create_body["handle"], handle);
+
+
+    let session_payload = json!({
+        "identifier": handle,
+        "password": password
+    });
+
+    let session_res = client.post(format!("{}/xrpc/com.atproto.server.createSession", BASE_URL))
+        .json(&session_payload)
+        .send()
+        .await
+        .expect("Failed to send createSession");
+
+    assert_eq!(session_res.status(), StatusCode::OK, "Failed to create session");
+    let session_body: Value = session_res.json().await.expect("createSession response was not JSON");
+
+    let session_jwt = session_body["accessJwt"].as_str().expect("Session response had no accessJwt").to_string();
+    assert_eq!(session_body["did"], new_did);
+
+
+    let profile_payload = json!({
+        "repo": new_did,
+        "collection": "app.bsky.actor.profile",
+        "rkey": "self", // The rkey for a profile is always "self"
+        "record": {
+            "$type": "app.bsky.actor.profile",
+            "displayName": "E2E Test User",
+            "description": "A user created by the e2e test suite."
+        }
+    });
+
+    let profile_res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(&session_jwt)
+        .json(&profile_payload)
+        .send()
+        .await
+        .expect("Failed to send putRecord for profile");
+
+    assert_eq!(profile_res.status(), StatusCode::OK, "Failed to create profile");
+
+
+    let params_get_profile = [
+        ("actor", &handle),
+    ];
+    let get_profile_res = client.get(format!("{}/xrpc/app.bsky.actor.getProfile", BASE_URL))
+        .query(&params_get_profile)
+        .send()
+        .await
+        .expect("Failed to send getProfile");
+
+    assert_eq!(get_profile_res.status(), StatusCode::OK, "getProfile did not return 200");
+    let profile_body: Value = get_profile_res.json().await.expect("getProfile response was not JSON");
+
+    assert_eq!(profile_body["did"], new_did);
+    assert_eq!(profile_body["handle"], handle);
+    assert_eq!(profile_body["displayName"], "E2E Test User");
+
+
+    let logout_res = client.post(format!("{}/xrpc/com.atproto.server.deleteSession", BASE_URL))
+        .bearer_auth(&session_jwt)
+        .send()
+        .await
+        .expect("Failed to send deleteSession");
+
+    assert_eq!(logout_res.status(), StatusCode::OK, "Failed to delete session");
+
+
+    let get_session_res = client.get(format!("{}/xrpc/com.atproto.server.getSession", BASE_URL))
+        .bearer_auth(&session_jwt)
+        .send()
+        .await
+        .expect("Failed to send getSession");
+
+    assert_eq!(get_session_res.status(), StatusCode::UNAUTHORIZED, "Session was still valid after logout");
+}
+
+async fn setup_new_user(handle_prefix: &str) -> (String, String) {
+    let client = client();
+    let ts = Utc::now().timestamp_millis();
+    let handle = format!("{}-{}.test", handle_prefix, ts);
+    let email = format!("{}-{}@test.com", handle_prefix, ts);
+    let password = "e2e-password-123";
+
+    let create_account_payload = json!({
+        "handle": handle,
+        "email": email,
+        "password": password
+    });
+    let create_res = client.post(format!("{}/xrpc/com.atproto.server.createAccount", BASE_URL))
+        .json(&create_account_payload)
+        .send()
+        .await
+        .expect("setup_new_user: Failed to send createAccount");
+    assert_eq!(create_res.status(), StatusCode::OK, "setup_new_user: Failed to create account");
+    let create_body: Value = create_res.json().await.expect("setup_new_user: createAccount response was not JSON");
+
+    let new_did = create_body["did"].as_str().expect("setup_new_user: Response had no DID").to_string();
+    let new_jwt = create_body["accessJwt"].as_str().expect("setup_new_user: Response had no accessJwt").to_string();
+
+    let profile_payload = json!({
+        "repo": new_did.clone(),
+        "collection": "app.bsky.actor.profile",
+        "rkey": "self",
+        "record": {
+            "$type": "app.bsky.actor.profile",
+            "displayName": format!("E2E User {}", handle),
+            "description": "A user created by the e2e test suite."
+        }
+    });
+    let profile_res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(&new_jwt)
+        .json(&profile_payload)
+        .send()
+        .await
+        .expect("setup_new_user: Failed to send putRecord for profile");
+    assert_eq!(profile_res.status(), StatusCode::OK, "setup_new_user: Failed to create profile");
+
+    (new_did, new_jwt)
+}
+
+async fn create_record_as(
+    client: &Client,
+    jwt: &str,
+    did: &str,
+    collection: &str,
+    record: Value,
+) -> (String, String) {
+    let payload = json!({
+        "repo": did,
+        "collection": collection,
+        "record": record
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.createRecord", BASE_URL))
+        .bearer_auth(jwt)
+        .json(&payload)
+        .send()
+        .await
+        .expect("create_record_as: Failed to send createRecord");
+
+    assert_eq!(res.status(), StatusCode::OK, "create_record_as: Failed to create record");
+    let body: Value = res.json().await.expect("create_record_as: response was not JSON");
+
+    let uri = body["uri"].as_str().expect("create_record_as: Response had no URI").to_string();
+    let cid = body["cid"].as_str().expect("create_record_as: Response had no CID").to_string();
+    (uri, cid)
+}
+
+async fn delete_record_as(
+    client: &Client,
+    jwt: &str,
+    did: &str,
+    collection: &str,
+    rkey: &str,
+) {
+    let payload = json!({
+        "repo": did,
+        "collection": collection,
+        "rkey": rkey
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", BASE_URL))
+        .bearer_auth(jwt)
+        .json(&payload)
+        .send()
+        .await
+        .expect("delete_record_as: Failed to send deleteRecord");
+
+    assert_eq!(res.status(), StatusCode::OK, "delete_record_as: Failed to delete record");
+}
+
+
+#[tokio::test]
+async fn test_notification_lifecycle() {
+    let client = client();
+
+    let (user_a_did, user_a_jwt) = setup_new_user("user-a-notif").await;
+    let (user_b_did, user_b_jwt) = setup_new_user("user-b-notif").await;
+
+    let (post_uri, post_cid) = create_record_as(
+        &client,
+        &user_a_jwt,
+        &user_a_did,
+        "app.bsky.feed.post",
+        json!({
+            "$type": "app.bsky.feed.post",
+            "text": "A post to be notified about",
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+    let post_ref = json!({ "uri": post_uri, "cid": post_cid });
+
+    let count_res_1 = client.get(format!("{}/xrpc/app.bsky.notification.getUnreadCount", BASE_URL))
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getUnreadCount 1 failed");
+    let count_body_1: Value = count_res_1.json().await.expect("count 1 not json");
+    assert_eq!(count_body_1["count"], 0, "Initial unread count was not 0");
+
+    create_record_as(
+        &client, &user_b_jwt, &user_b_did,
+        "app.bsky.graph.follow",
+        json!({
+            "$type": "app.bsky.graph.follow",
+            "subject": user_a_did,
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+    create_record_as(
+        &client, &user_b_jwt, &user_b_did,
+        "app.bsky.feed.like",
+        json!({
+            "$type": "app.bsky.feed.like",
+            "subject": post_ref,
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+    create_record_as(
+        &client, &user_b_jwt, &user_b_did,
+        "app.bsky.feed.post",
+        json!({
+            "$type": "app.bsky.feed.post",
+            "text": "This is a reply!",
+            "reply": { "root": post_ref.clone(), "parent": post_ref.clone() },
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let count_res_2 = client.get(format!("{}/xrpc/app.bsky.notification.getUnreadCount", BASE_URL))
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getUnreadCount 2 failed");
+    let count_body_2: Value = count_res_2.json().await.expect("count 2 not json");
+    assert_eq!(count_body_2["count"], 3, "Unread count was not 3 after actions");
+
+    let list_res = client.get(format!("{}/xrpc/app.bsky.notification.listNotifications", BASE_URL))
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("listNotifications failed");
+    let list_body: Value = list_res.json().await.expect("list not json");
+
+    let notifs = list_body["notifications"].as_array().expect("notifications not array");
+    assert_eq!(notifs.len(), 3, "Notification list did not have 3 items");
+
+    let has_follow = notifs.iter().any(|n| n["reason"] == "follow" && n["author"]["did"] == user_b_did);
+    let has_like = notifs.iter().any(|n| n["reason"] == "like" && n["author"]["did"] == user_b_did);
+    let has_reply = notifs.iter().any(|n| n["reason"] == "reply" && n["author"]["did"] == user_b_did);
+
+    assert!(has_follow, "Notification list missing 'follow'");
+    assert!(has_like, "Notification list missing 'like'");
+    assert!(has_reply, "Notification list missing 'reply'");
+
+    let count_res_3 = client.get(format!("{}/xrpc/app.bsky.notification.getUnreadCount", BASE_URL))
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getUnreadCount 3 failed");
+    let count_body_3: Value = count_res_3.json().await.expect("count 3 not json");
+    assert_eq!(count_body_3["count"], 0, "Unread count was not 0 after list");
+}
+
+
+#[tokio::test]
+async fn test_mute_lifecycle_filters_feed() {
+    let client = client();
+
+    let (user_a_did, user_a_jwt) = setup_new_user("user-a-mute").await;
+    let (user_b_did, user_b_jwt) = setup_new_user("user-b-mute").await;
+
+    let (post_uri, _) = create_record_as(
+        &client,
+        &user_b_jwt,
+        &user_b_did,
+        "app.bsky.feed.post",
+        json!({
+            "$type": "app.bsky.feed.post",
+            "text": "A post from User B",
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+
+    let feed_params_1 = [("actor", &user_b_did)];
+    let feed_res_1 = client.get(format!("{}/xrpc/app.bsky.feed.getAuthorFeed", BASE_URL))
+        .query(&feed_params_1)
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getAuthorFeed 1 failed");
+    let feed_body_1: Value = feed_res_1.json().await.expect("feed 1 not json");
+
+    let feed_1 = feed_body_1["feed"].as_array().expect("feed 1 not array");
+    let found_post_1 = feed_1.iter().any(|p| p["post"]["uri"] == post_uri);
+    assert!(found_post_1, "User B's post was not in their feed before mute");
+
+    let (mute_uri, _) = create_record_as(
+        &client, &user_a_jwt, &user_a_did,
+        "app.bsky.graph.mute",
+        json!({
+            "$type": "app.bsky.graph.mute",
+            "subject": user_b_did,
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+    let mute_rkey = mute_uri.split('/').last().unwrap();
+
+    let feed_params_2 = [("actor", &user_b_did)];
+    let feed_res_2 = client.get(format!("{}/xrpc/app.bsky.feed.getAuthorFeed", BASE_URL))
+        .query(&feed_params_2)
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getAuthorFeed 2 failed");
+    let feed_body_2: Value = feed_res_2.json().await.expect("feed 2 not json");
+
+    let feed_2 = feed_body_2["feed"].as_array().expect("feed 2 not array");
+    assert!(feed_2.is_empty(), "User B's feed was not empty after mute");
+
+    delete_record_as(
+        &client, &user_a_jwt, &user_a_did,
+        "app.bsky.graph.mute",
+        mute_rkey,
+    ).await;
+
+    let feed_params_3 = [("actor", &user_b_did)];
+    let feed_res_3 = client.get(format!("{}/xrpc/app.bsky.feed.getAuthorFeed", BASE_URL))
+        .query(&feed_params_3)
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getAuthorFeed 3 failed");
+    let feed_body_3: Value = feed_res_3.json().await.expect("feed 3 not json");
+
+    let feed_3 = feed_body_3["feed"].as_array().expect("feed 3 not array");
+    let found_post_3 = feed_3.iter().any(|p| p["post"]["uri"] == post_uri);
+    assert!(found_post_3, "User B's post did not reappear after unmute");
+}
+
+
+#[tokio::test]
+async fn test_record_update_conflict_lifecycle() {
+    let client = client();
+
+    let (user_did, user_jwt) = setup_new_user("user-conflict").await;
+
+    let get_res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&[
+            ("repo", &user_did),
+            ("collection", &"app.bsky.actor.profile".to_string()),
+            ("rkey", &"self".to_string()),
+        ])
+        .send().await.expect("getRecord failed");
+    let get_body: Value = get_res.json().await.expect("getRecord not json");
+    let cid_v1 = get_body["cid"].as_str().expect("Profile v1 had no CID").to_string();
+
+    let update_payload_v2 = json!({
+        "repo": user_did,
+        "collection": "app.bsky.actor.profile",
+        "rkey": "self",
+        "record": {
+            "$type": "app.bsky.actor.profile",
+            "displayName": "Updated Name (v2)"
+        },
+        "swapCommit": cid_v1 // <-- Correctly point to v1
+    });
+    let update_res_v2 = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(&user_jwt)
+        .json(&update_payload_v2)
+        .send().await.expect("putRecord v2 failed");
+    assert_eq!(update_res_v2.status(), StatusCode::OK, "v2 update failed");
+    let update_body_v2: Value = update_res_v2.json().await.expect("v2 body not json");
+    let cid_v2 = update_body_v2["cid"].as_str().expect("v2 response had no CID").to_string();
+
+    let update_payload_v3_stale = json!({
+        "repo": user_did,
+        "collection": "app.bsky.actor.profile",
+        "rkey": "self",
+        "record": {
+            "$type": "app.bsky.actor.profile",
+            "displayName": "Stale Update (v3)"
+        },
+        "swapCommit": cid_v1
+    });
+    let update_res_v3_stale = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(&user_jwt)
+        .json(&update_payload_v3_stale)
+        .send().await.expect("putRecord v3 (stale) failed");
+
+    assert_eq!(
+        update_res_v3_stale.status(),
+        StatusCode::CONFLICT,
+        "Stale update did not cause a 409 Conflict"
+    );
+
+    let update_payload_v3_good = json!({
+        "repo": user_did,
+        "collection": "app.bsky.actor.profile",
+        "rkey": "self",
+        "record": {
+            "$type": "app.bsky.actor.profile",
+            "displayName": "Good Update (v3)"
+        },
+        "swapCommit": cid_v2 // <-- Correct
+    });
+    let update_res_v3_good = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(&user_jwt)
+        .json(&update_payload_v3_good)
+        .send().await.expect("putRecord v3 (good) failed");
+
+    assert_eq!(update_res_v3_good.status(), StatusCode::OK, "v3 (good) update failed");
+}
+
+
+#[tokio::test]
+async fn test_complex_thread_deletion_lifecycle() {
+    let client = client();
+
+    let (user_a_did, user_a_jwt) = setup_new_user("user-a-thread").await;
+    let (user_b_did, user_b_jwt) = setup_new_user("user-b-thread").await;
+    let (user_c_did, user_c_jwt) = setup_new_user("user-c-thread").await;
+
+    let (p1_uri, p1_cid) = create_record_as(
+        &client, &user_a_jwt, &user_a_did,
+        "app.bsky.feed.post",
+        json!({
+            "$type": "app.bsky.feed.post",
+            "text": "P1 (Root)",
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+    let p1_ref = json!({ "uri": p1_uri.clone(), "cid": p1_cid.clone() });
+
+    let (p2_uri, p2_cid) = create_record_as(
+        &client, &user_b_jwt, &user_b_did,
+        "app.bsky.feed.post",
+        json!({
+            "$type": "app.bsky.feed.post",
+            "text": "P2 (Reply)",
+            "reply": { "root": p1_ref.clone(), "parent": p1_ref.clone() },
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+    let p2_ref = json!({ "uri": p2_uri.clone(), "cid": p2_cid.clone() });
+    let p2_rkey = p2_uri.split('/').last().unwrap().to_string();
+
+    let (p3_uri, _) = create_record_as(
+        &client, &user_c_jwt, &user_c_did,
+        "app.bsky.feed.post",
+        json!({
+            "$type": "app.bsky.feed.post",
+            "text": "P3 (Grandchild)",
+            "reply": { "root": p1_ref.clone(), "parent": p2_ref.clone() },
+            "createdAt": Utc::now().to_rfc3339()
+        }),
+    ).await;
+
+    let thread_res_1 = client.get(format!("{}/xrpc/app.bsky.feed.getPostThread", BASE_URL))
+        .query(&[("uri", &p1_uri)])
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getThread 1 failed");
+    let thread_body_1: Value = thread_res_1.json().await.expect("thread 1 not json");
+
+    let p1_replies = thread_body_1["thread"]["replies"].as_array().unwrap();
+    assert_eq!(p1_replies.len(), 1, "P1 should have 1 reply");
+    assert_eq!(p1_replies[0]["post"]["uri"], p2_uri, "P1's reply is not P2");
+
+    let p2_replies = p1_replies[0]["replies"].as_array().unwrap();
+    assert_eq!(p2_replies.len(), 1, "P2 should have 1 reply");
+    assert_eq!(p2_replies[0]["post"]["uri"], p3_uri, "P2's reply is not P3");
+
+    delete_record_as(
+        &client, &user_b_jwt, &user_b_did,
+        "app.bsky.feed.post",
+        &p2_rkey,
+    ).await;
+
+    let thread_res_2 = client.get(format!("{}/xrpc/app.bsky.feed.getPostThread", BASE_URL))
+        .query(&[("uri", &p1_uri)])
+        .bearer_auth(&user_a_jwt)
+        .send().await.expect("getThread 2 failed");
+    let thread_body_2: Value = thread_res_2.json().await.expect("thread 2 not json");
+
+    let p1_replies_2 = thread_body_2["thread"]["replies"].as_array().unwrap();
+    assert_eq!(p1_replies_2.len(), 1, "P1 should still have 1 reply (the deleted one)");
+
+    let deleted_post = &p1_replies_2[0];
+    assert_eq!(
+        deleted_post["$type"], "app.bsky.feed.defs#notFoundPost",
+        "P2 did not appear as a notFoundPost"
+    );
+    assert_eq!(deleted_post["uri"], p2_uri, "notFoundPost URI does not match P2");
+
+    let p3_reply = deleted_post["replies"].as_array().unwrap();
+    assert_eq!(p3_reply.len(), 1, "notFoundPost should still have P3 as a reply");
+    assert_eq!(p3_reply[0]["post"]["uri"], p3_uri, "The reply to the deleted post is not P3");
+}

tests/notification.rs

@@ -0,0 +1,31 @@
+mod common;
+use common::*;
+use reqwest::StatusCode;
+
+#[tokio::test]
+async fn test_list_notifications() {
+    let client = client();
+    let params = [
+        ("limit", "30"),
+    ];
+    let res = client.get(format!("{}/xrpc/app.bsky.notification.listNotifications", BASE_URL))
+        .query(&params)
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_get_unread_count() {
+    let client = client();
+    let res = client.get(format!("{}/xrpc/app.bsky.notification.getUnreadCount", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}

tests/repo.rs

@@ -0,0 +1,354 @@
+mod common;
+use common::*;
+
+use reqwest::{header, StatusCode};
+use serde_json::{json, Value};
+use chrono::Utc;
+
+#[tokio::test]
+#[ignore]
+async fn test_get_record() {
+    let client = client();
+    let params = [
+        ("repo", "did:plc:12345"),
+        ("collection", "app.bsky.actor.profile"),
+        ("rkey", "self"),
+    ];
+
+    let res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert_eq!(body["value"]["$type"], "app.bsky.actor.profile");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_get_record_not_found() {
+    let client = client();
+    let params = [
+        ("repo", "did:plc:12345"),
+        ("collection", "app.bsky.feed.post"),
+        ("rkey", "nonexistent"),
+    ];
+
+    let res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::NOT_FOUND);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert_eq!(body["error"], "NotFound");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_upload_blob_no_auth() {
+    let client = client();
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", BASE_URL))
+        .header(header::CONTENT_TYPE, "text/plain")
+        .body("no auth")
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert_eq!(body["error"], "AuthenticationFailed");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_upload_blob_success() {
+    let client = client();
+    let (token, _) = create_account_and_login(&client).await;
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", BASE_URL))
+        .header(header::CONTENT_TYPE, "text/plain")
+        .bearer_auth(token)
+        .body("This is our blob data")
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert!(body["blob"]["ref"]["$link"].as_str().is_some());
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_put_record_no_auth() {
+    let client = client();
+    let payload = json!({
+        "repo": "did:plc:123",
+        "collection": "app.bsky.feed.post",
+        "rkey": "fake",
+        "record": {}
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert_eq!(body["error"], "AuthenticationFailed");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_put_record_success() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let now = Utc::now().to_rfc3339();
+    let payload = json!({
+        "repo": did,
+        "collection": "app.bsky.feed.post",
+        "rkey": "e2e_test_post",
+        "record": {
+            "$type": "app.bsky.feed.post",
+            "text": "Hello from the e2e test script!",
+            "createdAt": now
+        }
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(token)
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert!(body.get("uri").is_some());
+    assert!(body.get("cid").is_some());
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_get_record_missing_params() {
+    let client = client();
+    // Missing `collection` and `rkey`
+    let params = [
+        ("repo", "did:plc:12345"),
+    ];
+
+    let res = client.get(format!("{}/xrpc/com.atproto.repo.getRecord", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    // This will fail (get 404) until the handler validates query params
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST, "Expected 400 for missing params");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_upload_blob_bad_token() {
+    let client = client();
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", BASE_URL))
+        .header(header::CONTENT_TYPE, "text/plain")
+        .bearer_auth(BAD_AUTH_TOKEN)
+        .body("This is our blob data")
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    // This *should* pass if the auth stub is working correctly
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert_eq!(body["error"], "AuthenticationFailed");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_put_record_mismatched_repo() {
+    let client = client();
+    let (token, _) = create_account_and_login(&client).await;
+    let now = Utc::now().to_rfc3339();
+    let payload = json!({
+        "repo": "did:plc:OTHER-USER", // This does NOT match AUTH_DID
+        "collection": "app.bsky.feed.post",
+        "rkey": "e2e_test_post",
+        "record": {
+            "$type": "app.bsky.feed.post",
+            "text": "Hello from the e2e test script!",
+            "createdAt": now
+        }
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(token)
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    // This will fail (get 200) until handler validates repo matches auth
+    assert_eq!(res.status(), StatusCode::FORBIDDEN, "Expected 403 for mismatched repo and auth");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_put_record_invalid_schema() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let now = Utc::now().to_rfc3339();
+    let payload = json!({
+        "repo": did,
+        "collection": "app.bsky.feed.post",
+        "rkey": "e2e_test_invalid",
+        "record": {
+            "$type": "app.bsky.feed.post",
+            // "text" field is missing, this is invalid
+            "createdAt": now
+        }
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.putRecord", BASE_URL))
+        .bearer_auth(token)
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    // This will fail (get 200) until handler validates record schema
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST, "Expected 400 for invalid record schema");
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_upload_blob_unsupported_mime_type() {
+    let client = client();
+    let (token, _) = create_account_and_login(&client).await;
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.uploadBlob", BASE_URL))
+        .header(header::CONTENT_TYPE, "application/xml")
+        .bearer_auth(token)
+        .body("<xml>not an image</xml>")
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    // This will fail (get 200) until handler validates mime type
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST, "Expected 400 for unsupported mime type");
+}
+
+#[tokio::test]
+async fn test_list_records() {
+    let client = client();
+    let (_, did) = create_account_and_login(&client).await;
+    let params = [
+        ("repo", did.as_str()),
+        ("collection", "app.bsky.feed.post"),
+        ("limit", "10"),
+    ];
+    let res = client.get(format!("{}/xrpc/com.atproto.repo.listRecords", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_delete_record() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let payload = json!({
+        "repo": did,
+        "collection": "app.bsky.feed.post",
+        "rkey": "some_post_to_delete"
+    });
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", BASE_URL))
+        .bearer_auth(token)
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_describe_repo() {
+    let client = client();
+    let (_, did) = create_account_and_login(&client).await;
+    let params = [
+        ("repo", did.as_str()),
+    ];
+    let res = client.get(format!("{}/xrpc/com.atproto.repo.describeRepo", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_create_record_success_with_generated_rkey() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let payload = json!({
+        "repo": did,
+        "collection": "app.bsky.feed.post",
+        "record": {
+            "$type": "app.bsky.feed.post",
+            "text": "Hello, world!",
+            "createdAt": "2025-12-02T12:00:00Z"
+        }
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.createRecord", BASE_URL))
+        .json(&payload)
+        .bearer_auth(token) // Assuming auth is required
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    let uri = body["uri"].as_str().unwrap();
+    assert!(uri.starts_with(&format!("at://{}/app.bsky.feed.post/", did)));
+    // assert_eq!(body["cid"], "bafyreihy"); // CID is now real
+}
+
+#[tokio::test]
+async fn test_create_record_success_with_provided_rkey() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+    let rkey = "custom-rkey";
+    let payload = json!({
+        "repo": did,
+        "collection": "app.bsky.feed.post",
+        "rkey": rkey,
+        "record": {
+            "$type": "app.bsky.feed.post",
+            "text": "Hello, world!",
+            "createdAt": "2025-12-02T12:00:00Z"
+        }
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.repo.createRecord", BASE_URL))
+        .json(&payload)
+        .bearer_auth(token) // Assuming auth is required
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert_eq!(body["uri"], format!("at://{}/app.bsky.feed.post/{}", did, rkey));
+    // assert_eq!(body["cid"], "bafyreihy"); // CID is now real
+}

tests/server.rs

@@ -0,0 +1,164 @@
+mod common;
+use common::*;
+
+use reqwest::StatusCode;
+use serde_json::{json, Value};
+
+#[tokio::test]
+async fn test_health() {
+    let client = client();
+    let res = client.get(format!("{}/health", BASE_URL))
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    assert_eq!(res.text().await.unwrap(), "OK");
+}
+
+#[tokio::test]
+async fn test_describe_server() {
+    let client = client();
+    let res = client.get(format!("{}/xrpc/com.atproto.server.describeServer", BASE_URL))
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert!(body.get("availableUserDomains").is_some());
+}
+
+#[tokio::test]
+async fn test_create_session() {
+    let client = client();
+
+    let handle = format!("user_{}", uuid::Uuid::new_v4());
+    let payload = json!({
+        "handle": handle,
+        "email": format!("{}@example.com", handle),
+        "password": "password"
+    });
+    let _ = client.post(format!("{}/xrpc/com.atproto.server.createAccount", BASE_URL))
+        .json(&payload)
+        .send()
+        .await;
+
+    let payload = json!({
+        "identifier": handle,
+        "password": "password"
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.server.createSession", BASE_URL))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Response was not valid JSON");
+    assert!(body.get("accessJwt").is_some());
+}
+
+#[tokio::test]
+async fn test_create_session_missing_identifier() {
+    let client = client();
+    let payload = json!({
+        "password": "password"
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.server.createSession", BASE_URL))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert!(res.status() == StatusCode::BAD_REQUEST || res.status() == StatusCode::UNPROCESSABLE_ENTITY,
+        "Expected 400 or 422 for missing identifier, got {}", res.status());
+}
+
+#[tokio::test]
+async fn test_create_account_invalid_handle() {
+    let client = client();
+    let payload = json!({
+        "handle": "invalid!handle.com",
+        "email": "test@example.com",
+        "password": "password"
+    });
+
+    let res = client.post(format!("{}/xrpc/com.atproto.server.createAccount", BASE_URL))
+        .json(&payload)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::BAD_REQUEST, "Expected 400 for invalid handle chars");
+}
+
+#[tokio::test]
+async fn test_get_session() {
+    let client = client();
+    let res = client.get(format!("{}/xrpc/com.atproto.server.getSession", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+}
+
+#[tokio::test]
+async fn test_refresh_session() {
+    let client = client();
+
+    let handle = format!("refresh_user_{}", uuid::Uuid::new_v4());
+    let payload = json!({
+        "handle": handle,
+        "email": format!("{}@example.com", handle),
+        "password": "password"
+    });
+    let _ = client.post(format!("{}/xrpc/com.atproto.server.createAccount", BASE_URL))
+        .json(&payload)
+        .send()
+        .await;
+
+    let login_payload = json!({
+        "identifier": handle,
+        "password": "password"
+    });
+    let res = client.post(format!("{}/xrpc/com.atproto.server.createSession", BASE_URL))
+        .json(&login_payload)
+        .send()
+        .await
+        .expect("Failed to login");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    let refresh_jwt = body["refreshJwt"].as_str().expect("No refreshJwt").to_string();
+    let access_jwt = body["accessJwt"].as_str().expect("No accessJwt").to_string();
+
+    let res = client.post(format!("{}/xrpc/com.atproto.server.refreshSession", BASE_URL))
+        .bearer_auth(&refresh_jwt)
+        .send()
+        .await
+        .expect("Failed to refresh");
+
+    assert_eq!(res.status(), StatusCode::OK);
+    let body: Value = res.json().await.expect("Invalid JSON");
+    assert!(body["accessJwt"].as_str().is_some());
+    assert!(body["refreshJwt"].as_str().is_some());
+    assert_ne!(body["accessJwt"].as_str().unwrap(), access_jwt);
+    assert_ne!(body["refreshJwt"].as_str().unwrap(), refresh_jwt);
+}
+
+#[tokio::test]
+async fn test_delete_session() {
+    let client = client();
+    let res = client.post(format!("{}/xrpc/com.atproto.server.deleteSession", BASE_URL))
+        .bearer_auth(AUTH_TOKEN)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
+}

tests/sync.rs

@@ -0,0 +1,34 @@
+mod common;
+use common::*;
+use reqwest::StatusCode;
+
+#[tokio::test]
+async fn test_get_repo() {
+    let client = client();
+    let params = [
+        ("did", AUTH_DID),
+    ];
+    let res = client.get(format!("{}/xrpc/com.atproto.sync.getRepo", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_get_blocks() {
+    let client = client();
+    let params = [
+        ("did", AUTH_DID),
+        // "cids" would be a list of CIDs
+    ];
+    let res = client.get(format!("{}/xrpc/com.atproto.sync.getBlocks", BASE_URL))
+        .query(&params)
+        .send()
+        .await
+        .expect("Failed to send request");
+
+    assert_eq!(res.status(), StatusCode::OK);
+}