don't
feat(knot): add /events endpoint
Signed-off-by: tjh <did:plc:65gha4t3avpfpzmvpbwovss7>
+73
-6
+14
crates/knot/src/model/knot_state.rs
+14
crates/knot/src/model/knot_state.rs
···
25
25
26
26
use super::config::KnotConfiguration;
27
27
28
+
#[derive(Clone)]
29
+
pub enum Event {
30
+
RefUpdate(lexicon::sh::tangled::git::RefUpdate<'static>),
31
+
}
32
+
28
33
#[derive(Debug)]
29
34
pub struct KnotState {
30
35
config: KnotConfiguration,
···
52
47
53
48
/// Thread pool for running synchronous tasks.
54
49
pool: ThreadPool,
50
+
51
+
events: tokio::sync::broadcast::Sender<Event>,
55
52
56
53
/// Stores JWT claims to prevent re-use.
57
54
jwt_claims: Mutex<HashMap<Box<str>, jwt::Claims>>,
···
81
74
.build()
82
75
.expect("Failed to build thread pool");
83
76
77
+
let (events, _) = tokio::sync::broadcast::channel(8);
78
+
84
79
let inner = Arc::new(Self {
85
80
config,
86
81
public_http,
···
90
81
jetstream,
91
82
store: database,
92
83
pool,
84
+
events,
93
85
jwt_claims: Default::default(),
94
86
repo_cache: Default::default(),
95
87
push_seed: Default::default(),
···
139
129
#[inline]
140
130
pub(crate) fn pool(&self) -> &ThreadPool {
141
131
&self.pool
132
+
}
133
+
134
+
pub(crate) fn subscribe_events(&self) -> tokio::sync::broadcast::Receiver<Event> {
135
+
self.events.subscribe()
142
136
}
143
137
144
138
/// Return a reference to the database shim.
+2
crates/knot/src/public.rs
+2
crates/knot/src/public.rs
+51
crates/knot/src/public/events.rs
+51
crates/knot/src/public/events.rs
···
1
+
use std::time::Duration;
2
+
3
+
use axum::{
4
+
extract::{
5
+
State, WebSocketUpgrade,
6
+
ws::{Message, WebSocket},
7
+
},
8
+
response::IntoResponse,
9
+
};
10
+
use tokio::time::Instant;
11
+
12
+
use crate::model::{Knot, knot_state::Event};
13
+
14
+
const KEEP_ALIVE: Duration = Duration::from_secs(45);
15
+
16
+
pub async fn handler(State(state): State<Knot>, ws: WebSocketUpgrade) -> impl IntoResponse {
17
+
ws.on_upgrade(|socket| handle_socket(state, socket))
18
+
}
19
+
20
+
async fn handle_socket(state: Knot, mut socket: WebSocket) {
21
+
let mut keep_alive = tokio::time::interval(KEEP_ALIVE);
22
+
let mut events = state.subscribe_events();
23
+
let start = Instant::now();
24
+
25
+
tracing::info!("new events subscriber");
26
+
loop {
27
+
let event = tokio::select! {
28
+
now = keep_alive.tick() => {
29
+
let bytes = (now.duration_since(start)).as_secs().to_string().into();
30
+
if let Err(error) = socket.send(Message::Ping(bytes)).await {
31
+
tracing::error!(?error, "failed to send ping");
32
+
break;
33
+
}
34
+
continue;
35
+
}
36
+
Ok(event) = events.recv() => {
37
+
event
38
+
}
39
+
};
40
+
41
+
match event {
42
+
Event::RefUpdate(ref_update) => {
43
+
let bytes = serde_json::to_string(&ref_update).unwrap();
44
+
if let Err(error) = socket.send(Message::Text(bytes.into())).await {
45
+
tracing::error!(?error, "failed to send ref update");
46
+
break;
47
+
}
48
+
}
49
+
}
50
+
}
51
+
}
+6
-6
crates/lexicon/src/sh/tangled/git.rs
+6
-6
crates/lexicon/src/sh/tangled/git.rs
···
6
6
/// `sh.tangled.git.refUpdate` record
7
7
///
8
8
/// See: <https://tangled.org/tangled.org/core/blob/master/lexicons/git/refUpdate.json>
9
-
#[derive(Debug, PartialEq, Eq, Deserialize, Serialize)]
9
+
#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
10
10
#[serde(rename_all = "camelCase")]
11
11
pub struct RefUpdate<'a> {
12
12
/// Ref being updated.
···
31
31
pub meta: Meta<'a>,
32
32
}
33
33
34
-
#[derive(Debug, PartialEq, Eq, Deserialize, Serialize)]
34
+
#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
35
35
#[serde(rename_all = "camelCase")]
36
36
pub struct Meta<'a> {
37
37
pub is_default_ref: bool,
···
42
42
pub commit_count: CommitCountBreakdown<'a>,
43
43
}
44
44
45
-
#[derive(Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
45
+
#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
46
46
#[serde(rename_all = "camelCase")]
47
47
pub struct LanguageBreakdown<'a> {
48
48
#[serde(borrow, default, skip_serializing_if = "Vec::is_empty")]
49
49
pub inputs: Vec<Language<'a>>,
50
50
}
51
51
52
-
#[derive(Debug, PartialEq, Eq, Deserialize, Serialize)]
52
+
#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
53
53
#[serde(rename_all = "camelCase")]
54
54
pub struct Language<'a> {
55
55
#[serde(borrow)]
···
57
57
pub size: u64,
58
58
}
59
59
60
-
#[derive(Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
60
+
#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize, Serialize)]
61
61
#[serde(rename_all = "camelCase")]
62
62
pub struct CommitCountBreakdown<'a> {
63
63
#[serde(borrow, default, skip_serializing_if = "Vec::is_empty")]
64
64
pub by_email: Vec<CommitCount<'a>>,
65
65
}
66
66
67
-
#[derive(Debug, PartialEq, Eq, Deserialize, Serialize)]
67
+
#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
68
68
#[serde(rename_all = "camelCase")]
69
69
pub struct CommitCount<'a> {
70
70
#[serde(borrow)]