Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
sqlite get/put pretty much works
mutex-wrapping the storage accidentally serializes all access which oops but also whatever
author
phil
date
(Sep 5, 2025, 5:30 PM -0400)
commit
b0a66a10
b0a66a102261d0b4e8a90d34cec3421073a7b728
parent
69d717a0
69d717a0be1e71e9ffcd63f1754125380999f482
+160
-26
+1
pocket/.gitignore
+1
pocket/.gitignore
···
1
+
prefs.sqlite3*
+2
pocket/src/lib.rs
+2
pocket/src/lib.rs
+29
-3
pocket/src/main.rs
+29
-3
pocket/src/main.rs
···
1
-
use pocket::serve;
1
+
use clap::Parser;
2
+
use pocket::{Storage, serve};
3
+
use std::path::PathBuf;
4
+
5
+
/// Slingshot record edge cache
6
+
#[derive(Parser, Debug, Clone)]
7
+
#[command(version, about, long_about = None)]
8
+
struct Args {
9
+
/// path to the sqlite db file
10
+
#[arg(long)]
11
+
db: Option<PathBuf>,
12
+
/// just initialize the db and exit
13
+
#[arg(long, action)]
14
+
init_db: bool,
15
+
/// the domain for serving a did doc (unused if running behind reflector)
16
+
#[arg(long)]
17
+
domain: Option<String>,
18
+
}
2
19
3
20
#[tokio::main]
4
21
async fn main() {
5
22
tracing_subscriber::fmt::init();
6
-
println!("Hello, world!");
7
-
serve("mac.cinnebar-tet.ts.net").await
23
+
log::info!("👖 hi");
24
+
let args = Args::parse();
25
+
let domain = args.domain.unwrap_or("bad-example.com".into());
26
+
let db_path = args.db.unwrap_or("prefs.sqlite3".into());
27
+
if args.init_db {
28
+
Storage::init(&db_path).unwrap();
29
+
log::info!("👖 initialized db at {db_path:?}. bye")
30
+
} else {
31
+
let storage = Storage::connect(db_path).unwrap();
32
+
serve(&domain, storage).await
33
+
}
8
34
}
+78
-23
pocket/src/server.rs
+78
-23
pocket/src/server.rs
···
1
-
use crate::TokenVerifier;
1
+
use crate::{Storage, TokenVerifier};
2
2
use poem::{
3
3
Endpoint, EndpointExt, Route, Server,
4
4
endpoint::{StaticFileEndpoint, make_sync},
5
5
http::Method,
6
6
listener::TcpListener,
7
-
middleware::{CatchPanic, Cors, SizeLimit, Tracing},
7
+
middleware::{CatchPanic, Cors, Tracing},
8
8
};
9
9
use poem_openapi::{
10
10
ApiResponse, ContactObject, ExternalDocumentObject, Object, OpenApi, OpenApiService,
···
15
15
};
16
16
use serde::Serialize;
17
17
use serde_json::{Value, json};
18
+
use std::sync::{Arc, Mutex};
18
19
19
20
#[derive(Debug, SecurityScheme)]
20
21
#[oai(ty = "bearer")]
···
51
52
})
52
53
}
53
54
54
-
#[derive(Object)]
55
+
#[derive(Debug, Object)]
55
56
#[oai(example = true)]
56
-
struct GetBskyPrefsResponseObject {
57
+
struct BskyPrefsObject {
57
58
/// at-uri for this record
58
59
preferences: Value,
59
60
}
60
-
impl Example for GetBskyPrefsResponseObject {
61
+
impl Example for BskyPrefsObject {
61
62
fn example() -> Self {
62
63
Self {
63
64
preferences: json!({
···
71
72
enum GetBskyPrefsResponse {
72
73
/// Record found
73
74
#[oai(status = 200)]
74
-
Ok(Json<GetBskyPrefsResponseObject>),
75
+
Ok(Json<BskyPrefsObject>),
75
76
/// Bad request or no preferences to return
76
77
#[oai(status = 400)]
77
78
BadRequest(XrpcError),
···
92
93
93
94
struct Xrpc {
94
95
verifier: TokenVerifier,
96
+
storage: Arc<Mutex<Storage>>,
95
97
}
96
98
97
99
#[OpenApi]
···
114
116
Err(e) => return GetBskyPrefsResponse::BadRequest(xrpc_error("boooo", e.to_string())),
115
117
};
116
118
log::info!("verified did: {did}/{aud}");
117
-
// TODO: fetch from storage
118
-
GetBskyPrefsResponse::Ok(Json(GetBskyPrefsResponseObject::example()))
119
+
120
+
let storage = self.storage.clone();
121
+
122
+
let Ok(Ok(res)) = tokio::task::spawn_blocking(move || {
123
+
storage
124
+
.lock()
125
+
.unwrap()
126
+
.get(&did, &aud)
127
+
.inspect_err(|e| log::error!("failed to get prefs: {e}"))
128
+
})
129
+
.await
130
+
else {
131
+
return GetBskyPrefsResponse::BadRequest(xrpc_error("boooo", "failed to get from db"));
132
+
};
133
+
134
+
let Some(serialized) = res else {
135
+
return GetBskyPrefsResponse::BadRequest(xrpc_error(
136
+
"NotFound",
137
+
"could not find prefs for u",
138
+
));
139
+
};
140
+
141
+
let preferences = match serde_json::from_str(&serialized) {
142
+
Ok(v) => v,
143
+
Err(e) => {
144
+
log::error!("failed to deserialize prefs: {e}");
145
+
return GetBskyPrefsResponse::BadRequest(xrpc_error(
146
+
"boooo",
147
+
"failed to deserialize prefs",
148
+
));
149
+
}
150
+
};
151
+
152
+
GetBskyPrefsResponse::Ok(Json(BskyPrefsObject { preferences }))
119
153
}
120
154
121
155
/// com.bad-example.pocket.putPreferences
···
129
163
async fn pocket_put_prefs(
130
164
&self,
131
165
XrpcAuth(auth): XrpcAuth,
132
-
Json(prefs): Json<Value>,
166
+
Json(prefs): Json<BskyPrefsObject>,
133
167
) -> PutBskyPrefsResponse {
134
168
let (did, aud) = match self
135
169
.verifier
···
141
175
};
142
176
log::info!("verified did: {did}/{aud}");
143
177
log::warn!("received prefs: {prefs:?}");
144
-
// TODO: put prefs into storage
145
-
PutBskyPrefsResponse::Ok(PlainText("hiiiiii".to_string()))
178
+
179
+
let storage = self.storage.clone();
180
+
let serialized = prefs.preferences.to_string();
181
+
182
+
let Ok(Ok(())) = tokio::task::spawn_blocking(move || {
183
+
storage
184
+
.lock()
185
+
.unwrap()
186
+
.put(&did, &aud, &serialized)
187
+
.inspect_err(|e| log::error!("failed to insert prefs: {e}"))
188
+
})
189
+
.await
190
+
else {
191
+
return PutBskyPrefsResponse::BadRequest(xrpc_error("boooo", "failed to put to db"));
192
+
};
193
+
194
+
PutBskyPrefsResponse::Ok(PlainText("saved.".to_string()))
146
195
}
147
196
}
148
197
···
178
227
make_sync(move |_| doc.clone())
179
228
}
180
229
181
-
pub async fn serve(domain: &str) -> () {
230
+
pub async fn serve(domain: &str, storage: Storage) -> () {
182
231
let verifier = TokenVerifier::default();
183
-
let api_service = OpenApiService::new(Xrpc { verifier }, "Pocket", env!("CARGO_PKG_VERSION"))
184
-
.server(domain)
185
-
.url_prefix("/xrpc")
186
-
.contact(
187
-
ContactObject::new()
188
-
.name("@microcosm.blue")
189
-
.url("https://bsky.app/profile/microcosm.blue"),
190
-
)
191
-
.description(include_str!("../api-description.md"))
192
-
.external_document(ExternalDocumentObject::new("https://microcosm.blue/pocket"));
232
+
let api_service = OpenApiService::new(
233
+
Xrpc {
234
+
verifier,
235
+
storage: Arc::new(Mutex::new(storage)),
236
+
},
237
+
"Pocket",
238
+
env!("CARGO_PKG_VERSION"),
239
+
)
240
+
.server(domain)
241
+
.url_prefix("/xrpc")
242
+
.contact(
243
+
ContactObject::new()
244
+
.name("@microcosm.blue")
245
+
.url("https://bsky.app/profile/microcosm.blue"),
246
+
)
247
+
.description(include_str!("../api-description.md"))
248
+
.external_document(ExternalDocumentObject::new("https://microcosm.blue/pocket"));
193
249
194
250
let app = Route::new()
195
251
.nest("/openapi", api_service.spec_endpoint())
196
252
.nest("/xrpc/", api_service)
197
253
.at("/.well-known/did.json", get_did_doc(domain))
198
254
.at("/", StaticFileEndpoint::new("./static/index.html"))
199
-
.with(SizeLimit::new(100 * 2_usize.pow(10)))
200
255
.with(
201
256
Cors::new()
202
257
.allow_method(Method::GET)
+50
pocket/src/storage.rs
+50
pocket/src/storage.rs
···
1
+
use rusqlite::{Connection, OptionalExtension, Result};
2
+
use std::path::Path;
3
+
4
+
pub struct Storage {
5
+
con: Connection,
6
+
}
7
+
8
+
impl Storage {
9
+
pub fn connect(path: impl AsRef<Path>) -> Result<Self> {
10
+
let con = Connection::open(path)?;
11
+
con.pragma_update(None, "journal_mode", "WAL")?;
12
+
con.pragma_update(None, "synchronous", "NORMAL")?;
13
+
con.pragma_update(None, "busy_timeout", "100")?;
14
+
con.pragma_update(None, "foreign_keys", "ON")?;
15
+
Ok(Self { con })
16
+
}
17
+
pub fn init(path: impl AsRef<Path>) -> Result<Self> {
18
+
let me = Self::connect(path)?;
19
+
me.con.execute(
20
+
r#"
21
+
create table prefs (
22
+
actor text not null,
23
+
aud text not null,
24
+
pref text not null,
25
+
primary key (actor, aud)
26
+
) strict"#,
27
+
(),
28
+
)?;
29
+
Ok(me)
30
+
}
31
+
pub fn put(&self, actor: &str, aud: &str, pref: &str) -> Result<()> {
32
+
self.con.execute(
33
+
r#"insert into prefs (actor, aud, pref)
34
+
values (?1, ?2, ?3)
35
+
on conflict do update set pref = excluded.pref"#,
36
+
[actor, aud, pref],
37
+
)?;
38
+
Ok(())
39
+
}
40
+
pub fn get(&self, actor: &str, aud: &str) -> Result<Option<String>> {
41
+
self.con
42
+
.query_one(
43
+
r#"select pref from prefs
44
+
where actor = ?1 and aud = ?2"#,
45
+
[actor, aud],
46
+
|row| row.get(0),
47
+
)
48
+
.optional()
49
+
}
50
+
}