don't
refactor: tidy
Signed-off-by: tjh <x@tjh.dev>
+54
-84
+31
-9
crates/gordian-knot/src/cli/serve.rs
+31
-9
crates/gordian-knot/src/cli/serve.rs
···
1
-
use std::{env, ffi, net::ToSocketAddrs as _, path, time::Duration};
1
+
use std::{env, ffi, io, net::ToSocketAddrs as _, path, time::Duration};
2
2
3
3
use anyhow::Context as _;
4
4
use axum::http::{Request, Response};
···
202
202
.build()
203
203
.expect("Failed to build runtime");
204
204
205
+
unsafe { env::set_var("GIT_CONFIG_GLOBAL", &self.git_config) };
206
+
205
207
runtime.block_on(serve_knot(self))
206
208
}
207
209
}
208
210
209
211
pub async fn serve_knot(arguments: Serve) -> anyhow::Result<()> {
210
-
unsafe { env::set_var("GIT_CONFIG_GLOBAL", &arguments.git_config) };
211
-
212
212
let tempdir = tempfile::TempDir::with_prefix("gordian-knot-")?;
213
213
let hooks_path = if let Some(path) = &arguments.hooks {
214
214
// @TODO Verify hooks exist in the specified path.
···
295
295
let shutdown = CancellationToken::new();
296
296
297
297
// Spawn the internal API.
298
-
tasks.spawn(gordian_knot::serve_all(
298
+
tasks.spawn(serve_all(
299
299
gordian_knot::private::router()
300
300
.layer(
301
301
ServiceBuilder::new()
···
339
339
)
340
340
.with_state(knot);
341
341
342
-
tasks.spawn(gordian_knot::serve_all(
343
-
router,
344
-
public_listeners,
345
-
shutdown.child_token(),
346
-
));
342
+
tasks.spawn(serve_all(router, public_listeners, shutdown.child_token()));
347
343
348
344
tasks.spawn(wait_for_shutdown(shutdown));
349
345
···
435
439
status => tracing::info!(?status, ?latency),
436
440
}
437
441
}
442
+
}
443
+
444
+
pub async fn serve_all(
445
+
router: axum::Router,
446
+
listeners: impl IntoIterator<Item = TcpListener>,
447
+
shutdown: CancellationToken,
448
+
) -> io::Result<()> {
449
+
let mut service = JoinSet::new();
450
+
for listener in listeners {
451
+
let router = router.clone();
452
+
let addr = listener.local_addr()?;
453
+
tracing::info!(?addr, "listening on socket");
454
+
455
+
let shutdown = shutdown.child_token();
456
+
service.spawn(async move {
457
+
axum::serve(listener, router)
458
+
.with_graceful_shutdown(async move { shutdown.cancelled().await })
459
+
.await
460
+
});
461
+
}
462
+
463
+
for task in service.join_all().await {
464
+
task?;
465
+
}
466
+
467
+
Ok(())
438
468
}
+17
-30
crates/gordian-knot/src/lib.rs
+17
-30
crates/gordian-knot/src/lib.rs
···
1
-
use std::io;
2
-
3
-
use axum::Router;
4
1
use gordian_types::Nsid;
5
-
use tokio::{net::TcpListener, task::JoinSet};
6
-
use tokio_util::sync::CancellationToken;
7
2
8
3
pub mod command;
9
4
pub mod extractors;
···
34
39
35
40
pub use gordian_lexicon as lexicon;
36
41
42
+
/// Atmosphere lexicon for the records relevant to knot server.
43
+
#[derive(Debug, serde::Deserialize, serde::Serialize)]
44
+
#[serde(tag = "$type")]
45
+
pub enum Lexicon<'a> {
46
+
#[serde(rename = "sh.tangled.knot")]
47
+
Knot(lexicon::sh_tangled::knot::Knot),
48
+
#[serde(borrow, rename = "sh.tangled.knot.member")]
49
+
KnotMember(lexicon::sh_tangled::knot::Member<'a>),
50
+
#[serde(rename = "sh.tangled.publicKey")]
51
+
PublicKey(lexicon::sh_tangled::PublicKey<'a>),
52
+
#[serde(rename = "sh.tangled.repo")]
53
+
Repo(lexicon::sh_tangled::repo::Repo<'a>),
54
+
#[serde(rename = "sh.tangled.repo.collaborator")]
55
+
RepoCollaborator(lexicon::sh_tangled::repo::Collaborator<'a>),
56
+
}
57
+
37
58
/// NSIDs of interest to a knot server.
38
59
pub const NSIDS: &[&Nsid] = {
39
60
&[
···
60
49
]
61
50
};
62
51
63
-
pub async fn serve_all(
64
-
router: Router,
65
-
listeners: impl IntoIterator<Item = TcpListener>,
66
-
shutdown: CancellationToken,
67
-
) -> io::Result<()> {
68
-
let mut service = JoinSet::new();
69
-
for listener in listeners {
70
-
let router = router.clone();
71
-
let addr = listener.local_addr()?;
72
-
tracing::info!(?addr, "listening on socket");
73
-
74
-
let shutdown = shutdown.child_token();
75
-
service.spawn(async move {
76
-
axum::serve(listener, router)
77
-
.with_graceful_shutdown(async move { shutdown.cancelled().await })
78
-
.await
79
-
});
80
-
}
81
-
82
-
for task in service.join_all().await {
83
-
task?;
84
-
}
85
-
86
-
Ok(())
87
-
}
52
+
pub use model::Knot;
88
53
89
54
#[cfg(test)]
90
55
mod tests {
+2
crates/gordian-knot/src/public.rs
+2
crates/gordian-knot/src/public.rs
+4
-4
crates/gordian-knot/src/services/jetstream.rs
+4
-4
crates/gordian-knot/src/services/jetstream.rs
···
1
1
use crate::{
2
-
lexicon::Lexicon,
3
-
model::{Knot, KnotState},
2
+
Lexicon,
3
+
model::Knot,
4
4
services::rbac::{
5
5
Action, AddCollaboratorPolicy, AddMemberPolicy, Policy, PolicyResult::*,
6
6
RemoveCollaboratorPolicy, RemoveMemberPolicy, RepositoryCreatePolicy,
···
146
146
#[tracing::instrument(skip(knot))]
147
147
async fn process_public_key<'db, 'c, 'k>(
148
148
_: &JetstreamClient,
149
-
knot: &KnotState,
149
+
knot: &Knot,
150
150
event: &'c CommitEvent<'c>,
151
151
) -> anyhow::Result<()>
152
152
where
···
182
182
#[tracing::instrument(skip(knot))]
183
183
async fn process_repo(
184
184
_: &JetstreamClient,
185
-
knot: &KnotState,
185
+
knot: &Knot,
186
186
event: &CommitEvent<'_>,
187
187
) -> anyhow::Result<()> {
188
188
match event {
-41
crates/gordian-lexicon/src/lib.rs
-41
crates/gordian-lexicon/src/lib.rs
···
10
10
/// Types which aren't part of the lexicon, but are used for serializing
11
11
/// and deserializing.
12
12
pub mod extra;
13
-
14
-
#[derive(Debug, serde::Deserialize, serde::Serialize)]
15
-
#[serde(tag = "$type")]
16
-
pub enum Lexicon<'a> {
17
-
#[serde(borrow, rename = "sh.tangled.actor.profile")]
18
-
ActorProfile(sh_tangled::actor::Profile<'a>),
19
-
#[serde(rename = "sh.tangled.feed.reaction")]
20
-
FeedReaction(sh_tangled::feed::Reaction<'a>),
21
-
#[serde(rename = "sh.tangled.feed.star")]
22
-
FeedStar(sh_tangled::feed::Star),
23
-
#[serde(rename = "sh.tangled.graph.follow")]
24
-
GraphFollow(sh_tangled::graph::Follow<'a>),
25
-
#[serde(rename = "sh.tangled.knot")]
26
-
Knot(sh_tangled::knot::Knot),
27
-
#[serde(rename = "sh.tangled.knot.member")]
28
-
KnotMember(sh_tangled::knot::Member<'a>),
29
-
#[serde(rename = "sh.tangled.publicKey")]
30
-
PublicKey(sh_tangled::PublicKey<'a>),
31
-
#[serde(rename = "sh.tangled.repo")]
32
-
Repo(sh_tangled::repo::Repo<'a>),
33
-
#[serde(rename = "sh.tangled.repo.issue")]
34
-
RepoIssue(sh_tangled::repo::Issue<'a>),
35
-
#[serde(rename = "sh.tangled.repo.issue.comment")]
36
-
RepoIssueComment(sh_tangled::repo::issue::Comment<'a>),
37
-
#[serde(rename = "sh.tangled.repo.issue.state")]
38
-
RepoIssueState(sh_tangled::repo::issue::State<'a>),
39
-
#[serde(rename = "sh.tangled.repo.collaborator")]
40
-
RepoCollaborator(sh_tangled::repo::Collaborator<'a>),
41
-
#[serde(rename = "sh.tangled.repo.pull")]
42
-
RepoPull(sh_tangled::repo::Pull<'a>),
43
-
#[serde(rename = "sh.tangled.repo.pull.comment")]
44
-
RepoPullComment(sh_tangled::repo::pull::Comment<'a>),
45
-
#[serde(rename = "sh.tangled.repo.pull.status")]
46
-
RepoPullstate(sh_tangled::repo::pull::Status<'a>),
47
-
#[serde(rename = "sh.tangled.spindle")]
48
-
Spindle(sh_tangled::spindle::Spindle),
49
-
#[serde(rename = "sh.tangled.spindle.member")]
50
-
SpindleMember(sh_tangled::spindle::Member<'a>),
51
-
#[serde(rename = "sh.tangled.string")]
52
-
String(sh_tangled::string::String<'a>),
53
-
}