don't
feat(knot): cleanup startup and naming
Signed-off-by: tjh <x@tjh.dev>
+293
-289
+8
crates/jetstream/src/client_config.rs
+8
crates/jetstream/src/client_config.rs
+48
-80
crates/knot/src/cli.rs
+48
-80
crates/knot/src/cli.rs
···
1
-
use anyhow::Context as _;
2
1
use atproto::did::OwnedDid;
3
2
use clap::{ArgAction, Parser, ValueEnum, ValueHint};
4
3
use gix::bstr::BString;
5
-
use identity::{HttpClient, ResolveError, Resolver};
6
4
use knot::model::config::{DEFAULT_READMES, KnotConfiguration, RepoCacheConfig};
7
-
use std::{env, path::PathBuf, str::FromStr, time::Duration};
8
-
use tower::ServiceBuilder;
9
-
10
-
const USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"));
5
+
use std::{env, path::PathBuf, time::Duration};
6
+
use url::Url;
11
7
12
8
pub fn parse() -> Arguments {
13
9
Arguments::parse()
14
10
}
15
11
16
-
#[derive(Debug, Parser)]
12
+
#[derive(Clone, Debug, Parser)]
17
13
#[command(about, author, version)]
18
14
pub struct Arguments {
19
15
/// FQDN of the knot.
···
19
23
20
24
/// Handle or DID of the knot owner.
21
25
#[arg(long, short, env = "KNOT_OWNER")]
22
-
pub owner: HandleOrDid,
26
+
pub owner: OwnedDid,
23
27
24
28
/// Base path for repositories.
25
29
#[arg(long, short, value_hint = ValueHint::DirPath, env = "KNOT_REPO_BASE")]
···
50
54
#[arg(default_value = "https://plc.directory")]
51
55
pub plc_directory: String,
52
56
57
+
#[arg(long, short, value_delimiter = ',', value_hint = ValueHint::Url, env = "KNOT_JETSREAM")]
58
+
#[arg(default_value = default_jetstream_instances())]
59
+
pub jetstream: Vec<Url>,
60
+
53
61
/// Acceptable authorization methods for git pushes over http.
54
62
#[arg(hide = true, long, require_equals = true, value_delimiter = ',')]
55
63
#[arg(env = "KNOT_AUTH_METHODS")]
···
83
83
pub repo_cache_live: u64,
84
84
}
85
85
86
-
impl Arguments {
87
-
pub async fn build(&self) -> anyhow::Result<(Resolver, HttpClient, KnotConfiguration)> {
88
-
let rate_limit: &'static _ = Box::leak(Box::new(
89
-
ServiceBuilder::new()
90
-
.layer(tower::buffer::BufferLayer::new(1))
91
-
.layer(tower::limit::RateLimitLayer::new(
92
-
10,
93
-
Duration::from_secs(10),
94
-
)),
95
-
));
86
+
#[derive(Debug, thiserror::Error)]
87
+
pub enum Error {
88
+
#[error("unable to build 'did:web:{{name}}' from knot fqdn: {0}")]
89
+
Name(#[from] atproto::did::Error),
90
+
}
96
91
97
-
let public_http = reqwest::ClientBuilder::new()
98
-
.connector_layer(rate_limit)
99
-
.timeout(Duration::from_secs(2))
100
-
.user_agent(USER_AGENT)
101
-
.http2_keep_alive_while_idle(true)
102
-
.https_only(true)
103
-
.build()
104
-
.context("Failed to build public HTTP client")?;
92
+
impl TryFrom<Arguments> for KnotConfiguration {
93
+
type Error = Error;
94
+
fn try_from(value: Arguments) -> Result<KnotConfiguration, Self::Error> {
95
+
let Arguments {
96
+
name,
97
+
owner,
98
+
repos: repo_path,
99
+
hooks: hook_path,
100
+
git_config,
101
+
bind: _,
102
+
db: _,
103
+
plc_directory: _,
104
+
jetstream: _,
105
+
auth_methods: _,
106
+
require_signed_push: _,
107
+
repo_cache_size,
108
+
repo_cache_idle,
109
+
repo_cache_live,
110
+
} = value;
105
111
106
-
let resolver = Resolver::builder()
107
-
.plc_directory(self.plc_directory.clone())
108
-
.build_with(public_http.clone());
112
+
// @TODO Validate?
109
113
110
-
let owner_did = self
111
-
.owner
112
-
.resolve_to_did(&resolver)
113
-
.await
114
-
.context("Failed to resolve owner handle")?;
114
+
let instance = format!("did:web:{name}").try_into()?;
115
115
116
-
let instance_audience = format!("did:web:{}", self.name).try_into()?;
117
-
118
-
let repo_cache = RepoCacheConfig {
119
-
size: self.repo_cache_size,
120
-
idle: Duration::from_secs(self.repo_cache_idle),
121
-
live: Duration::from_secs(self.repo_cache_live),
122
-
};
123
-
124
-
let config = KnotConfiguration {
125
-
owner_did,
126
-
instance_name: self.name.clone().into_boxed_str(),
127
-
instance_audience,
128
-
repo_path: self.repos.clone(),
129
-
hook_path: self.hooks.clone(),
130
-
git_config: self.git_config.clone(),
116
+
Ok(KnotConfiguration {
117
+
owner,
118
+
instance,
119
+
repo_path,
120
+
hook_path,
121
+
git_config,
131
122
readmes: DEFAULT_READMES
132
123
.iter()
133
124
.map(|v| BString::new(v.to_vec()))
134
125
.collect(),
135
-
repo_cache,
136
-
};
137
-
138
-
Ok((resolver, public_http, config))
126
+
repo_cache: RepoCacheConfig {
127
+
size: repo_cache_size,
128
+
idle: Duration::from_secs(repo_cache_idle),
129
+
live: Duration::from_secs(repo_cache_live),
130
+
},
131
+
})
139
132
}
140
133
}
141
134
···
142
149
env::current_dir().expect("current working directory should be readable")
143
150
}
144
151
145
-
#[derive(Clone, Debug)]
146
-
pub enum HandleOrDid {
147
-
Did(OwnedDid),
148
-
Handle(String),
149
-
}
150
-
151
-
impl HandleOrDid {
152
-
pub async fn resolve_to_did(&self, resolver: &Resolver) -> Result<OwnedDid, ResolveError> {
153
-
match self {
154
-
Self::Did(did) => Ok(did.clone()),
155
-
Self::Handle(handle) => {
156
-
let (did, _) = resolver.resolve(handle).await?;
157
-
tracing::info!(?handle, ?did, "resolved owner");
158
-
Ok(did)
159
-
}
160
-
}
161
-
}
162
-
}
163
-
164
-
impl FromStr for HandleOrDid {
165
-
type Err = anyhow::Error;
166
-
fn from_str(s: &str) -> Result<Self, Self::Err> {
167
-
match s.parse() {
168
-
Ok(did) => Ok(Self::Did(did)),
169
-
Err(_) => Ok(Self::Handle(s.to_string())),
170
-
}
171
-
}
152
+
fn default_jetstream_instances() -> String {
153
+
jetstream::PUBLIC_JETSTREAM_INSTANCES.join(",")
172
154
}
+110
-58
crates/knot/src/main.rs
+110
-58
crates/knot/src/main.rs
···
7
7
extract::Query,
8
8
http::{HeaderName, Request, Response},
9
9
};
10
-
use futures_util::StreamExt;
10
+
use futures_util::StreamExt as _;
11
+
use identity::Resolver;
11
12
use jetstream::client_config::JetstreamConfig;
12
13
use knot::{
13
-
model::{Knot, KnotState},
14
+
model::{Knot, KnotState, config::KnotConfiguration},
14
15
services::database::DataStore,
15
16
};
16
17
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
17
18
use std::{
19
+
borrow::Cow,
18
20
collections::BTreeMap,
19
21
env,
20
22
ffi::OsStr,
···
41
39
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
42
40
#[global_allocator]
43
41
static GLOBAL: Jemalloc = Jemalloc;
42
+
43
+
const USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"));
44
44
45
45
fn main() -> anyhow::Result<()> {
46
46
tracing_subscriber::registry()
···
122
118
}
123
119
)?);
124
120
125
-
let pool = {
126
-
let connect_options = SqliteConnectOptions::new()
127
-
.filename(&arguments.db)
128
-
.create_if_missing(true)
129
-
.foreign_keys(true)
130
-
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
121
+
let database = {
122
+
let pool = {
123
+
let connect_options = SqliteConnectOptions::new()
124
+
.filename(&arguments.db)
125
+
.create_if_missing(true)
126
+
.foreign_keys(true)
127
+
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
131
128
132
-
SqlitePoolOptions::new()
133
-
.connect_with(connect_options)
134
-
.await?
129
+
SqlitePoolOptions::new()
130
+
.connect_with(connect_options)
131
+
.await?
132
+
};
133
+
134
+
sqlx::migrate!().run(&pool).await?;
135
+
DataStore::new(pool)
135
136
};
136
-
137
-
// Run database migrations.
138
-
sqlx::migrate!().run(&pool).await?;
139
-
140
-
let db = DataStore::new(pool.clone());
141
-
142
-
let (resolver, public_http, config) = arguments.build().await?;
143
137
144
138
// Ensure the instance owner is also a member.
145
139
//
146
-
// This uses an invalid, empty rkey, and the maximum TID value as a revision ID to
147
-
// ensure this entry cannot be overwritten by an event from the firehose.
148
-
let needs_initial_fetch = db
140
+
// Use an invalid, empty rkey, and the maximum TID value as a revision ID to
141
+
// ensure this entry does not get overwritten by an event from the firehose.
142
+
let needs_initial_fetch = database
149
143
.upsert_knot_member(
150
144
"",
151
145
&Tid::MAX.to_string(),
152
146
"",
153
-
&config.owner_did,
147
+
&arguments.owner,
154
148
&arguments.name,
155
149
OffsetDateTime::now_utc(),
156
150
)
157
151
.await?;
158
152
159
-
let jetstream_config = {
160
-
let cursor = db
161
-
.get_jetstream_cursor()
153
+
let public_http = reqwest::ClientBuilder::new()
154
+
.timeout(Duration::from_secs(2))
155
+
.user_agent(USER_AGENT)
156
+
.http2_keep_alive_while_idle(true)
157
+
.https_only(true)
158
+
.build()
159
+
.context("Failed to build public HTTP client")?;
160
+
161
+
let resolver = Resolver::builder()
162
+
.plc_directory(arguments.plc_directory.clone())
163
+
.build_with(public_http.clone());
164
+
165
+
// Bind listeners for the public API.
166
+
let mut public_listeners = Vec::with_capacity(arguments.bind.len());
167
+
for addr in &arguments.bind {
168
+
for socket in addr.to_socket_addrs()? {
169
+
let listener = TcpListener::bind(socket).await?;
170
+
public_listeners.push(listener);
171
+
}
172
+
}
173
+
174
+
// Bind listeners for the private API.
175
+
let mut private_listeners = Vec::with_capacity(2);
176
+
for socket in "localhost:0".to_socket_addrs()? {
177
+
let listener = TcpListener::bind(socket).await?;
178
+
private_listeners.push(listener);
179
+
}
180
+
181
+
// The knot needs to know the sockets we've bound the private API.
182
+
let private_addrs = private_listeners
183
+
.iter()
184
+
.map(|listener| listener.local_addr())
185
+
.collect::<Result<Vec<_>, std::io::Error>>()?;
186
+
187
+
tracing::info!(?private_addrs, "bound internal API");
188
+
189
+
let jetstream_instances: Vec<_> = arguments
190
+
.jetstream
191
+
.iter()
192
+
.map(|url| Cow::Owned(url.to_string()))
193
+
.collect();
194
+
195
+
let config: KnotConfiguration = arguments.try_into()?;
196
+
let knot_state = KnotState::new(config, resolver, public_http, database, &private_addrs)?;
197
+
let knot = Knot::from(knot_state);
198
+
199
+
let mut service = JoinSet::new();
200
+
201
+
// Spawn the internal API.
202
+
service.spawn(knot::serve_all(
203
+
knot::private::router().with_state(knot.clone()),
204
+
private_listeners,
205
+
));
206
+
207
+
if needs_initial_fetch {
208
+
tracing::info!("Appears to be first run, fetching records");
209
+
knot.backfill_public_keys().await.unwrap();
210
+
knot.backfill_repositories(&knot.config.owner)
162
211
.await
163
-
.context("Failed to query last jetstream cursor from datastore")?
212
+
.unwrap();
213
+
}
214
+
215
+
// Spawn the jetstream consumer.
216
+
{
217
+
let cursor = knot
218
+
.database()
219
+
.get_jetstream_cursor()
220
+
.await?
164
221
.map(|odt| (odt, (odt.unix_timestamp_nanos() / 1000).unsigned_abs()));
165
222
166
223
if let Some((cursor, cursor_us)) = &cursor {
···
229
164
}
230
165
231
166
let mut config = JetstreamConfig::default()
167
+
.with_instances(jetstream_instances)
232
168
.with_cursor(cursor.map(|(_, us)| us))
233
169
.with_collections([
234
170
"sh.tangled.knot.member",
···
238
172
"sh.tangled.repo.collaborator",
239
173
]);
240
174
241
-
let mut member_dids = db.members();
175
+
let mut member_dids = knot.database().members();
242
176
while let Some(did) = member_dids.next().await {
243
177
config
244
178
.subscriber_options
···
246
180
.expect("knot members shouldn't exceed maximum DID filters");
247
181
}
248
182
249
-
config
183
+
let (jetstream, jetstream_rx, jetstream_task) = config.connect();
184
+
let knot = knot.clone();
185
+
service.spawn(async move {
186
+
tokio::join!(knot::services::jetstream::jetstream_task(
187
+
jetstream,
188
+
knot.clone(),
189
+
jetstream_rx,
190
+
),);
191
+
192
+
panic!("jetstream consumer/task completed");
193
+
});
194
+
195
+
service.spawn(async move {
196
+
jetstream_task.await;
197
+
Ok(())
198
+
});
250
199
};
251
200
252
-
let mut service = JoinSet::new();
253
-
254
-
let (knot_state, knot_tasks) =
255
-
KnotState::new(config, resolver, public_http, jetstream_config, db)?;
256
-
let knot: Knot = knot_state.into();
257
-
258
-
if needs_initial_fetch {
259
-
tracing::info!("Appears to be first run, fetching records");
260
-
knot.backfill_public_keys().await.unwrap();
261
-
knot.backfill_repositories(&knot.config.owner_did)
262
-
.await
263
-
.unwrap();
264
-
}
265
-
201
+
// Build the public API.
266
202
let router = knot::public::router()
267
203
.layer(SetRequestIdLayer::new(
268
204
X_REQUEST_ID,
···
303
235
)
304
236
.with_state(knot);
305
237
306
-
let mut public_listeners = Vec::with_capacity(arguments.bind.len());
307
-
for addr in &arguments.bind {
308
-
for socket in addr.to_socket_addrs()? {
309
-
let listener = TcpListener::bind(socket).await?;
310
-
public_listeners.push(listener);
311
-
}
312
-
}
313
-
314
238
service.spawn(knot::serve_all(router, public_listeners));
315
-
316
-
service.spawn(async move {
317
-
tracing::debug!("starting knot tasks");
318
-
for _task in knot_tasks.join_all().await {
319
-
//
320
-
}
321
-
Ok(())
322
-
});
323
239
324
240
for task in service.join_all().await {
325
241
if let Err(error) = task {
+17
-17
crates/knot/src/model/config.rs
+17
-17
crates/knot/src/model/config.rs
···
1
1
//! Knot configuration.
2
2
use atproto::{Did, did::OwnedDid};
3
3
use gix::bstr::BString;
4
+
use rustc_hash::FxHashSet;
4
5
use std::{
5
-
collections::HashSet,
6
6
path::{Path, PathBuf},
7
7
time::Duration,
8
8
};
···
27
27
28
28
#[derive(Debug)]
29
29
pub struct KnotConfiguration {
30
-
pub owner_did: OwnedDid,
31
-
pub instance_name: Box<str>,
32
-
pub instance_audience: OwnedDid,
30
+
pub owner: OwnedDid,
31
+
pub instance: OwnedDid,
33
32
pub repo_path: PathBuf,
34
33
pub hook_path: PathBuf,
35
34
pub git_config: PathBuf,
36
-
pub readmes: HashSet<BString>,
35
+
pub readmes: FxHashSet<BString>,
37
36
pub repo_cache: RepoCacheConfig,
38
37
}
39
38
···
44
45
}
45
46
46
47
impl KnotConfiguration {
47
-
/// The FQDN of the knot-server.
48
48
#[inline]
49
-
pub const fn instance_name(&self) -> &str {
50
-
&self.instance_name
49
+
pub fn instance_ident(&self) -> &str {
50
+
self.instance().ident()
51
51
}
52
52
53
+
/// The DID of the knot.
53
54
#[inline]
54
-
pub fn instance_audience(&self) -> &Did {
55
-
&self.instance_audience
55
+
pub fn instance(&self) -> &Did {
56
+
assert_eq!(self.instance.method(), "web");
57
+
&self.instance
56
58
}
57
59
58
-
/// The DID of the knot-server owner.
60
+
/// The DID of the knot owner.
59
61
#[inline]
60
-
pub fn owner_did(&self) -> &Did {
61
-
&self.owner_did
62
+
pub fn owner(&self) -> &Did {
63
+
&self.owner
62
64
}
63
65
64
-
/// Base path to the knot-server's repositories.
66
+
/// Base path to the knot's repositories.
65
67
#[inline]
66
-
pub fn repository_path(&self) -> &Path {
68
+
pub fn repository_base(&self) -> &Path {
67
69
self.repo_path.as_path()
68
70
}
69
71
70
-
/// Path to the knot-server's global git hooks directory.
72
+
/// Path to the knot's global git hooks directory.
71
73
#[inline]
72
74
pub fn hook_path(&self) -> &Path {
73
75
self.hook_path.as_path()
···
79
79
self.git_config.as_path()
80
80
}
81
81
82
-
pub fn readmes(&self) -> &HashSet<BString> {
82
+
pub fn readmes(&self) -> &FxHashSet<BString> {
83
83
&self.readmes
84
84
}
85
85
}
+34
-70
crates/knot/src/model/knot_state.rs
+34
-70
crates/knot/src/model/knot_state.rs
···
1
1
use std::{
2
2
collections::HashMap,
3
3
io,
4
-
net::{TcpListener, ToSocketAddrs as _},
4
+
net::SocketAddr,
5
5
ops,
6
6
sync::{Arc, Mutex},
7
7
time::Duration,
···
11
11
use bytes::Bytes;
12
12
use futures_util::{FutureExt, future::BoxFuture};
13
13
use identity::{HttpClient, Resolver};
14
-
use jetstream::{JetstreamClient, client_config::JetstreamConfig};
15
14
use lexicon::sh::tangled::{git::RefUpdate, repo::Repo};
16
15
use moka::future::{Cache, CacheBuilder};
17
16
use rayon::{ThreadPool, ThreadPoolBuilder};
18
17
use serde::Serialize;
19
18
use time::OffsetDateTime;
20
-
use tokio::task::JoinSet;
21
19
use tokio_stream::StreamExt as _;
22
20
23
21
use crate::{
···
58
60
/// Reqwest client.
59
61
///
60
62
// @TODO Wrap this so prevent requests to sensitive/private endpoints.
61
-
public_http: HttpClient,
63
+
http: HttpClient,
62
64
63
-
/// Jetstream client.
64
-
pub jetstream: JetstreamClient,
65
-
66
-
store: DataStore,
65
+
database: DataStore,
67
66
68
67
/// Thread pool for running synchronous tasks.
69
68
pool: ThreadPool,
···
77
82
}
78
83
79
84
impl KnotState {
80
-
pub fn new(
85
+
pub fn new<'a>(
81
86
config: KnotConfiguration,
82
87
resolver: Resolver,
83
-
public_http: HttpClient,
84
-
jetstream_config: JetstreamConfig,
88
+
http: HttpClient,
85
89
database: DataStore,
86
-
) -> io::Result<(Arc<Self>, JoinSet<io::Result<()>>)> {
90
+
private_binds: impl IntoIterator<Item = &'a SocketAddr>,
91
+
) -> io::Result<Arc<Self>> {
87
92
let pool = ThreadPoolBuilder::new()
88
93
.build()
89
94
.expect("Failed to build thread pool");
90
95
91
96
let (events, _) = tokio::sync::broadcast::channel(16);
92
97
93
-
let (jetstream, jetstream_rx, jetstream_task) = jetstream_config.connect();
94
-
95
-
// Setup the private API.
96
-
let private_listeners = "localhost:0"
97
-
.to_socket_addrs()?
98
-
.map(|socket| {
99
-
let bound = TcpListener::bind(socket)?;
100
-
bound.set_nonblocking(true)?;
101
-
tokio::net::TcpListener::from_std(bound)
102
-
})
103
-
.collect::<Result<Vec<_>, io::Error>>()?;
104
-
105
-
let private_addrs = private_listeners
106
-
.iter()
107
-
.map(|listener| Ok(format!("http://{}/", listener.local_addr()?)))
108
-
.collect::<Result<Vec<_>, io::Error>>()?
98
+
let private_addrs = private_binds
99
+
.into_iter()
100
+
.map(|socket| format!("http://{socket}/"))
101
+
.collect::<Vec<_>>()
109
102
.join(" ");
110
-
111
-
tracing::info!(?private_addrs, "bound internal API");
112
103
113
104
let repo_cache = CacheBuilder::new(config.repo_cache.size)
114
105
.name("repository_cache")
···
105
124
106
125
let inner = Arc::new(Self {
107
126
config,
108
-
public_http,
127
+
http,
109
128
resolver,
110
-
jetstream,
111
-
store: database,
129
+
database,
112
130
pool,
113
131
events,
114
132
repo_key_cache: CacheBuilder::new(REPO_KEY_CACHE_SIZE)
···
119
139
private_addrs,
120
140
});
121
141
122
-
let mut tasks = JoinSet::new();
123
-
124
-
let _ = {
125
-
let state = Arc::clone(&inner);
126
-
tasks.spawn(async move {
127
-
tokio::join!(
128
-
crate::services::jetstream::jetstream_task(Arc::clone(&state), jetstream_rx,),
129
-
jetstream_task
130
-
);
131
-
132
-
panic!("jetstream consumer/task completed");
133
-
});
134
-
let internal = crate::private::router().with_state(Arc::clone(&inner).into());
135
-
tasks.spawn(crate::serve_all(internal, private_listeners))
136
-
};
137
-
138
-
Ok((inner, tasks))
142
+
Ok(inner)
139
143
}
140
144
141
145
/// Return a reference to the identity resolver.
···
146
182
}
147
183
}
148
184
149
-
/// Return a reference to the database shim.
185
+
/// Get a reference to the database.
150
186
#[inline]
151
-
pub fn store(&self) -> &DataStore {
152
-
&self.store
187
+
pub fn database(&self) -> &DataStore {
188
+
&self.database
153
189
}
154
190
155
191
pub async fn fetch_public_keys(&self, did: &Did) -> anyhow::Result<()> {
156
-
let db = self.store().clone();
192
+
let db = self.database().clone();
157
193
let did = did.to_owned();
158
194
let rev = Tid::MIN.to_string();
159
195
160
196
atrepo::fetch_collection::<_, anyhow::Error>(
161
197
self.resolver(),
162
-
&self.public_http,
198
+
&self.http,
163
199
&did.clone(),
164
200
"sh.tangled.publicKey",
165
201
async move |records| {
···
191
227
192
228
/// Fetch public keys for all knot members.
193
229
pub async fn backfill_public_keys(&self) -> anyhow::Result<()> {
194
-
let mut knot_members = self.store().knot_members();
230
+
let mut knot_members = self.database().knot_members();
195
231
while let Some(member) = knot_members.next().await {
196
232
let member = member?;
197
233
if let Err(error) = self.fetch_public_keys(&member).await {
···
206
242
}
207
243
208
244
pub async fn backfill_repositories(&self, did: &Did) -> anyhow::Result<()> {
209
-
let base = self.config.repository_path().to_path_buf();
210
-
let db = self.store().clone();
211
-
let instance = self.instance_name().to_owned();
245
+
let base = self.config.repository_base().to_path_buf();
246
+
let db = self.database().clone();
247
+
let instance = self.instance_ident().to_owned();
212
248
let did = did.to_owned();
213
249
let rev = Tid::MIN.to_string();
214
250
215
251
atrepo::fetch_collection::<_, anyhow::Error>(
216
252
self.resolver(),
217
-
&self.public_http,
253
+
&self.http,
218
254
&did.clone(),
219
255
"sh.tangled.repo",
220
256
async move |records| {
···
283
319
.service_endpoint;
284
320
285
321
let response = self
286
-
.public_http
322
+
.http
287
323
.get(get_record_url(pds.clone(), did, collection, rkey))
288
324
.send()
289
325
.await?
···
319
355
};
320
356
321
357
let (rkey, _) = self
322
-
.store()
358
+
.database()
323
359
.resolve_repository(&owner, repo_path.name())
324
360
.await?
325
361
.ok_or(RepositoryResolveError::NotFound)?;
···
347
383
.repo_cache
348
384
.try_get_with_by_ref(repo_key, async {
349
385
let path = self
350
-
.repository_path()
386
+
.repository_base()
351
387
.join(repo_key.owner_str())
352
388
.join(repo_key.rkey());
353
389
···
374
410
}
375
411
376
412
pub fn create_repo(&self, did: &Did, rkey: &str, name: &str) -> anyhow::Result<()> {
377
-
let path = self.repository_path().join(did.as_str()).join(rkey);
413
+
let path = self.repository_base().join(did.as_str()).join(rkey);
378
414
let repo = gix::init_bare(&path)?;
379
415
tracing::info!(?repo, "created repository");
380
416
381
417
// Create a symlink to map the repository name -> rkey.
382
-
let symlink_path = self.repository_path().join(did.as_str()).join(name);
418
+
let symlink_path = self.repository_base().join(did.as_str()).join(name);
383
419
let _ = std::fs::remove_file(&symlink_path);
384
420
std::os::unix::fs::symlink(rkey, &symlink_path)?;
385
421
···
387
423
}
388
424
389
425
pub fn delete_repo(&self, did: &Did, rkey: &str) -> anyhow::Result<()> {
390
-
let path = self.repository_path().join(did.as_str()).join(rkey);
426
+
let path = self.repository_base().join(did.as_str()).join(rkey);
391
427
392
-
let mut delete_path = self.repository_path().join("deleted");
428
+
let mut delete_path = self.repository_base().join("deleted");
393
429
let _ = std::fs::create_dir_all(&delete_path);
394
430
395
431
delete_path.push(format!("{}-{}", did, rkey));
···
440
476
now: i64,
441
477
) -> BoxFuture<'b, Result<Option<auth::jwt::Claims>, AuthorizationClaimsStoreError>> {
442
478
async move {
443
-
let claims = self.store().get_claims(jti, now).await.ok().flatten();
479
+
let claims = self.database().get_claims(jti, now).await.ok().flatten();
444
480
445
481
// If the claims have expired, remove them.
446
482
if matches!(&claims, Some(claims) if claims.exp < now) {
447
-
self.store()
483
+
self.database()
448
484
.delete_claims(jti)
449
485
.await
450
486
.map_err(|error| AuthorizationClaimsStoreError(error.into()))?;
···
463
499
now: i64,
464
500
) -> BoxFuture<'_, Result<(), AuthorizationClaimsStoreError>> {
465
501
async move {
466
-
self.store()
502
+
self.database()
467
503
.store_claims(claims, now)
468
504
.await
469
505
.map_err(|error| AuthorizationClaimsStoreError(error.into()))?;
+2
-1
crates/knot/src/model/repository.rs
+2
-1
crates/knot/src/model/repository.rs
···
8
8
};
9
9
use gix::bstr::{BString, ByteSlice};
10
10
use lexicon::sh::tangled::repo::{branch, get_default_branch, languages, refs, tree};
11
+
use rustc_hash::FxHashSet;
11
12
use serde::Deserialize;
12
13
13
14
use crate::{
···
382
381
pub fn tree(
383
382
&self,
384
383
params: tree::Input,
385
-
readmes: &HashSet<BString>,
384
+
readmes: &FxHashSet<BString>,
386
385
) -> XrpcResult<Json<tree::Output>> {
387
386
let (tip, immutable) = self.resolve_revspec(params.rev.as_deref())?;
388
387
let dotdot = params.path.clone().and_then(|mut path| {
+3
-3
crates/knot/src/private.rs
+3
-3
crates/knot/src/private.rs
···
246
246
.map_err(|error| (StatusCode::BAD_REQUEST, error.to_string()))?;
247
247
248
248
for public_key in state
249
-
.store()
249
+
.database()
250
250
.public_keys_for_did(&user_did)
251
251
.await
252
252
.map_err(|error| (StatusCode::INTERNAL_SERVER_ERROR, error.to_string()))?
···
292
292
// Our hook refers to the repository using DID and rkey, but Tangled needs DID and name, so we
293
293
// need to lookup the repository's name in the database.
294
294
let (_, repo_name) = state
295
-
.store()
295
+
.database()
296
296
.resolve_repository(&repo_key.owner, repo_key.rkey())
297
297
.await
298
298
.map_err(errors::RepoNotFound)?
···
334
334
let ts = OffsetDateTime::now_utc();
335
335
let event = Event::RefUpdate(Arc::new(ref_update));
336
336
let id = state
337
-
.store()
337
+
.database()
338
338
.insert_event(
339
339
ts,
340
340
&repo_key.owner,
+1
-1
crates/knot/src/public/events.rs
+1
-1
crates/knot/src/public/events.rs
···
66
66
67
67
let (mut sender, mut receiver) = socket.split();
68
68
69
-
let mut past_events = state.store().get_events(&start_ts);
69
+
let mut past_events = state.database().get_events(&start_ts);
70
70
while let Some(Ok(event)) = past_events.next().await {
71
71
cursor = event.id;
72
72
let wrapper = EventWrapper {
+1
-1
crates/knot/src/public/git/error.rs
+1
-1
crates/knot/src/public/git/error.rs
···
46
46
fn authenticate_header(knot: &Knot) -> HeaderValue {
47
47
// @TODO We should probably generate a challenge code for the client to sign.
48
48
// Inform the client which knot it needs to authenticate with.
49
-
HeaderValue::from_str(&format!("knot={}", knot.instance_name()))
49
+
HeaderValue::from_str(&format!("knot={}", knot.instance_ident()))
50
50
.expect("Header should be valid")
51
51
}
52
52
}
+1
-1
crates/knot/src/public/xrpc/sh/tangled.rs
+1
-1
crates/knot/src/public/xrpc/sh/tangled.rs
+3
-3
crates/knot/src/public/xrpc/sh/tangled/repo.rs
+3
-3
crates/knot/src/public/xrpc/sh/tangled/repo.rs
···
116
116
// Use the minimum rev value so *any* firehose-derived entry will have priority.
117
117
let rev = Tid::MIN.to_string();
118
118
let is_new = knot
119
-
.store()
119
+
.database()
120
120
.insert_repository(&claims.iss, ¶ms.rkey, &rev, "", &repo)
121
121
.await
122
122
.map_err(errors::RepoError)?;
123
123
124
-
if is_new && repo.knot == knot.instance_name() {
124
+
if is_new && repo.knot == knot.instance_ident() {
125
125
knot.create_repo(&claims.iss, ¶ms.rkey, &repo.name)
126
126
.map_err(errors::RepoError)?;
127
127
}
···
176
176
};
177
177
178
178
if let Some(record) = knot
179
-
.store()
179
+
.database()
180
180
.delete_repository(&delete)
181
181
.await
182
182
.map_err(errors::RepoError)?
+54
-43
crates/knot/src/services/jetstream.rs
+54
-43
crates/knot/src/services/jetstream.rs
···
1
1
use crate::{
2
-
model::KnotState,
2
+
model::{Knot, KnotState},
3
3
services::rbac::{
4
4
Action, AddCollaboratorPolicy, AddMemberPolicy, Policy, PolicyResult::*,
5
5
RemoveCollaboratorPolicy, RemoveMemberPolicy, RepositoryCreatePolicy,
6
6
RepositoryDeletePolicy, RepositoryRef,
7
7
},
8
8
};
9
-
use jetstream::{CommitEvent, Event};
9
+
use jetstream::{CommitEvent, Event, JetstreamClient};
10
10
use lexicon::Lexicon;
11
-
use std::{sync::Arc, time::Duration};
11
+
use std::time::Duration;
12
12
use tokio::time::Instant;
13
13
14
-
pub async fn jetstream_task(state: Arc<KnotState>, jetstream_rx: jetstream::JetstreamReceiver) {
14
+
pub async fn jetstream_task(
15
+
client: JetstreamClient,
16
+
knot: Knot,
17
+
jetstream_rx: jetstream::JetstreamReceiver,
18
+
) {
15
19
let mut last_jetstream_sync: Option<Instant> = None;
16
20
17
21
while let Some(event) = jetstream_rx.recv_async().await {
···
33
29
match &event {
34
30
Event::Commit(commit) => match commit.collection() {
35
31
"sh.tangled.knot.member" => {
36
-
if let Err(error) = process_knot_member(&state, commit).await {
32
+
if let Err(error) = process_knot_member(&client, &knot, commit).await {
37
33
tracing::error!(?error, "failed to process 'sh.tangled.knot_member' record")
38
34
}
39
35
}
40
36
"sh.tangled.publicKey" => {
41
-
if let Err(error) = process_public_key(&state, commit).await {
37
+
if let Err(error) = process_public_key(&client, &knot, commit).await {
42
38
tracing::error!(?error, "failed to process 'sh.tangled.publicKey' record")
43
39
}
44
40
}
45
41
"sh.tangled.repo" => {
46
-
if let Err(error) = process_repo(&state, commit).await {
42
+
if let Err(error) = process_repo(&client, &knot, commit).await {
47
43
tracing::error!(?error, "failed to process 'sh.tangled.repo' record")
48
44
}
49
45
}
50
46
"sh.tangled.repo.collaborator" => {
51
-
if let Err(error) = process_repo_collaborator(&state, commit).await {
47
+
if let Err(error) = process_repo_collaborator(&client, &knot, commit).await {
52
48
tracing::error!(
53
49
?error,
54
50
"failed to process 'sh.tangled.repo.collaborator' record"
···
58
54
collection => unimplemented!("no handler implemented for '{collection}' commits"),
59
55
},
60
56
Event::Account(account) => {
61
-
state.resolver().invalidate_did(account.did).await;
57
+
knot.resolver().invalidate_did(account.did).await;
62
58
}
63
59
Event::Identity(identity) => {
64
-
state.resolver().invalidate_did(identity.did).await;
60
+
knot.resolver().invalidate_did(identity.did).await;
65
61
}
66
62
}
67
63
68
64
if last_jetstream_sync.is_none_or(|value| value.elapsed() > Duration::from_secs(1)) {
69
-
match state.store().store_jetstream_cursor(event.ts()).await {
65
+
match knot.database().store_jetstream_cursor(event.ts()).await {
70
66
Ok(()) => last_jetstream_sync = Some(Instant::now()),
71
67
Err(error) => tracing::error!(?error, "failed to log jetstream event"),
72
68
};
···
77
73
}
78
74
79
75
async fn process_public_key<'db, 'c, 'k>(
80
-
state: &KnotState,
76
+
_client: &JetstreamClient,
77
+
knot: &KnotState,
81
78
event: &'c CommitEvent<'c>,
82
79
) -> anyhow::Result<()>
83
80
where
···
91
86
return Err(anyhow::anyhow!("expected a 'sh.tangled.publicKey' record"));
92
87
};
93
88
94
-
if !state.store().is_knot_member(event.did()).await? {
89
+
if !knot.database().is_knot_member(event.did()).await? {
95
90
return Ok(());
96
91
}
97
92
98
-
state
99
-
.store()
93
+
knot.database()
100
94
.upsert_public_key(commit.did, commit.rkey, commit.rev, commit.cid, &key)
101
95
.await?;
102
96
}
103
97
CommitEvent::Delete(delete) => {
104
98
assert_eq!(delete.collection, "sh.tangled.publicKey");
105
-
state
106
-
.store()
99
+
knot.database()
107
100
.delete_public_key(delete.did, delete.rkey, delete.rev)
108
101
.await?;
109
102
}
···
110
107
Ok(())
111
108
}
112
109
113
-
#[tracing::instrument(skip(state))]
114
-
async fn process_repo(state: &KnotState, event: &CommitEvent<'_>) -> anyhow::Result<()> {
110
+
#[tracing::instrument(skip(knot))]
111
+
async fn process_repo(
112
+
client: &JetstreamClient,
113
+
knot: &KnotState,
114
+
event: &CommitEvent<'_>,
115
+
) -> anyhow::Result<()> {
115
116
match event {
116
117
CommitEvent::Create(commit) => {
117
118
let Lexicon::Repo(repository) = serde_json::from_str(commit.record.get())? else {
118
119
return Err(anyhow::anyhow!("expected a 'sh.tangled.repo' record"));
119
120
};
120
121
121
-
if repository.knot != state.instance_name() {
122
+
if repository.knot != knot.instance_ident() {
122
123
tracing::debug!(
123
124
did = %event.did(),
124
125
rkey = %event.rkey(),
···
134
127
135
128
let policy = RepositoryCreatePolicy;
136
129
let can_create = policy
137
-
.evaluate_access(&commit.did, &Action::RepositoryCreate, state, state)
130
+
.evaluate_access(&commit.did, &Action::RepositoryCreate, knot, knot)
138
131
.await;
139
132
140
133
if !matches!(can_create, Granted) {
···
142
135
return Ok(());
143
136
}
144
137
145
-
let is_new = state
146
-
.store()
138
+
let is_new = knot
139
+
.database()
147
140
.insert_repository(commit.did, commit.rkey, commit.rev, commit.cid, &repository)
148
141
.await?;
149
142
···
157
150
return Ok(());
158
151
}
159
152
160
-
state.create_repo(commit.did, commit.rkey, &repository.name)?;
153
+
knot.create_repo(commit.did, commit.rkey, &repository.name)?;
161
154
}
162
155
CommitEvent::Update(commit) => {
163
156
let Lexicon::Repo(repository) = serde_json::from_str(commit.record.get())? else {
164
157
return Err(anyhow::anyhow!("expected a 'sh.tangled.repo' record"));
165
158
};
166
159
167
-
if repository.knot != state.instance_name() {
160
+
if repository.knot != knot.instance_ident() {
168
161
tracing::debug!(
169
162
did = %event.did(),
170
163
rkey = %event.rkey(),
···
176
169
177
170
// @TODO Does this need auth?
178
171
179
-
state
180
-
.store()
172
+
knot.database()
181
173
.update_repository(commit.did, commit.rkey, commit.rev, commit.cid, &repository)
182
174
.await?;
183
175
}
184
176
CommitEvent::Delete(commit) => {
185
177
// First determine whether the repository exists on this knot
186
-
if state
187
-
.store()
178
+
if knot
179
+
.database()
188
180
.resolve_repository(commit.did, commit.rkey)
189
181
.await?
190
182
.is_none()
···
199
193
let policy = RepositoryDeletePolicy;
200
194
let repository = RepositoryRef::new(commit.did, commit.rkey);
201
195
let can_create = policy
202
-
.evaluate_access(&commit.did, &Action::RepositoryDelete, &repository, state)
196
+
.evaluate_access(&commit.did, &Action::RepositoryDelete, &repository, knot)
203
197
.await;
204
198
205
199
if !matches!(can_create, Granted) {
···
207
201
return Ok(());
208
202
}
209
203
210
-
if let Some(record) = state.store().delete_repository(commit).await? {
211
-
state.delete_repo(&record.did, &record.rkey)?;
204
+
if let Some(record) = knot.database().delete_repository(commit).await? {
205
+
knot.delete_repo(&record.did, &record.rkey)?;
212
206
}
213
207
}
214
208
}
···
216
210
Ok(())
217
211
}
218
212
219
-
async fn process_knot_member(state: &KnotState, event: &CommitEvent<'_>) -> anyhow::Result<()> {
213
+
async fn process_knot_member(
214
+
client: &JetstreamClient,
215
+
state: &KnotState,
216
+
event: &CommitEvent<'_>,
217
+
) -> anyhow::Result<()> {
220
218
match event {
221
219
CommitEvent::Create(commit) | CommitEvent::Update(commit) => {
222
220
let policy = AddMemberPolicy;
···
239
229
));
240
230
};
241
231
242
-
if member.domain != state.instance_name() {
232
+
if member.domain != state.instance_ident() {
243
233
return Ok(());
244
234
}
245
235
246
236
if state
247
-
.store()
237
+
.database()
248
238
.upsert_knot_member(
249
239
commit.rkey,
250
240
commit.rev,
···
256
246
.await?
257
247
{
258
248
// Start tracking the DID of the new member.
259
-
state.jetstream.add_did(member.subject).await?;
249
+
client.add_did(member.subject).await?;
260
250
state.fetch_public_keys(member.subject).await?;
261
251
}
262
252
···
274
264
}
275
265
276
266
if let Some(did) = state
277
-
.store()
267
+
.database()
278
268
.delete_knot_member(delete.rkey, delete.rev)
279
269
.await?
280
270
{
281
271
// Stop tracking the DID of the new member.
282
-
state.jetstream.remove_did(did).await?;
272
+
client.remove_did(did).await?;
283
273
}
284
274
285
275
Ok(())
···
288
278
}
289
279
290
280
async fn process_repo_collaborator(
281
+
client: &JetstreamClient,
291
282
state: &KnotState,
292
283
event: &CommitEvent<'_>,
293
284
) -> anyhow::Result<()> {
···
320
309
}
321
310
322
311
if state
323
-
.store()
312
+
.database()
324
313
.upsert_repository_collaborator(
325
314
commit.did,
326
315
commit.rkey,
···
333
322
)
334
323
.await?
335
324
{
336
-
state.jetstream.add_did(coll.subject).await?;
325
+
client.add_did(coll.subject).await?;
337
326
state.fetch_public_keys(coll.subject).await?;
338
327
}
339
328
···
341
330
}
342
331
CommitEvent::Delete(delete) => {
343
332
let (repo_did, repo_rkey, _) = state
344
-
.store()
333
+
.database()
345
334
.get_repository_collaborator_record(delete.did, delete.rkey)
346
335
.await?;
347
336
···
362
351
}
363
352
364
353
if let Some(did) = state
365
-
.store()
354
+
.database()
366
355
.delete_repository_collaborator(delete.did, delete.rkey, delete.rev)
367
356
.await?
368
357
{
369
-
state.jetstream.remove_did(did).await?;
358
+
client.remove_did(did).await?;
370
359
}
371
360
372
361
Ok(())
+6
-6
crates/knot/src/services/rbac.rs
+6
-6
crates/knot/src/services/rbac.rs
···
77
77
async move {
78
78
let is_member = subject == resource.owner
79
79
|| context
80
-
.store()
80
+
.database()
81
81
.is_repository_member(&resource.owner, &resource.rkey, subject)
82
82
.await
83
83
.is_ok_and(|val| val);
···
102
102
_: &'a KnotState,
103
103
) -> BoxFuture<'a, PolicyResult> {
104
104
async move {
105
-
let is_owner = subject == resource.owner_did();
105
+
let is_owner = subject == resource.owner();
106
106
match (action, is_owner) {
107
107
(Action::KnotMemberCreate, true) => PolicyResult::Granted,
108
108
(_, _) => PolicyResult::Denied,
···
123
123
_: &'a KnotState,
124
124
) -> BoxFuture<'a, PolicyResult> {
125
125
async move {
126
-
let is_owner = subject == resource.owner_did();
126
+
let is_owner = subject == resource.owner();
127
127
match (action, is_owner) {
128
128
(Action::KnotMemberDelete, true) => PolicyResult::Granted,
129
129
(_, _) => PolicyResult::Denied,
···
188
188
context: &'a KnotState,
189
189
) -> BoxFuture<'a, PolicyResult> {
190
190
async move {
191
-
let is_member = subject == resource.owner_did()
191
+
let is_member = subject == resource.owner()
192
192
|| context
193
-
.store()
193
+
.database()
194
194
.is_knot_member(subject)
195
195
.await
196
196
.is_ok_and(|val| val);
···
215
215
context: &'a KnotState,
216
216
) -> BoxFuture<'a, PolicyResult> {
217
217
async move {
218
-
let is_owner = subject == context.owner_did() || resource.did == subject;
218
+
let is_owner = subject == context.owner() || resource.did == subject;
219
219
match (action, is_owner) {
220
220
(Action::RepositoryDelete, true) => PolicyResult::Granted,
221
221
(_, _) => PolicyResult::Denied,