netget/client/activitypub/
mod.rs

1//! ActivityPub client: one actor, served (document, key, inbox) on a local port by the
2//! server's instance code, driven by the handler. Verified inbox activities and the outcome
3//! of every action are events.
4pub mod actions;
5use crate::client::llm_budget::call_llm_for_client;
6use crate::llm::actions::client_trait::{Client, ClientActionResult};
7use crate::logging::emit::Log;
8use crate::protocol::{ConnectContext, Event};
9use crate::server::activitypub::{self as ap, instance::Instance};
10use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
11use crate::state::{AccessLogOwner, ClientStatus};
12pub use actions::ActivityPubClientProtocol;
13use anyhow::{Context, Result};
14use hyper::{server::conn::http1, service::service_fn};
15use hyper_util::rt::{TokioIo, TokioTimer};
16use serde_json::{json, Value};
17use std::convert::Infallible;
18use std::sync::Arc;
19use tokio::sync::mpsc;
20
21/// A handler chain (answer → event → answer …) stops after this many follow-ups.
22pub const MAX_FOLLOWUP_DEPTH: usize = 8;
23/// Connections the actor's endpoint serves at once.
24pub const MAX_ENDPOINT_CONNECTIONS: usize = 64;
25
26pub async fn connect(ctx: ConnectContext) -> Result<std::net::SocketAddr> {
27    let params = ctx.startup_params.as_ref();
28    let username = params
29        .map(|p| p.get_optional_string("username"))
30        .transpose()?
31        .flatten()
32        .unwrap_or_else(|| actions::DEFAULT_USERNAME.into());
33    let listen = params
34        .map(|p| p.get_optional_string("listen"))
35        .transpose()?
36        .flatten()
37        .unwrap_or_else(|| actions::DEFAULT_LISTEN.into());
38    let configured = params
39        .map(|p| p.get_optional_string("base_url"))
40        .transpose()?
41        .flatten();
42    let listener = tokio::net::TcpListener::bind(&listen)
43        .await
44        .with_context(|| format!("cannot serve the actor on {listen}"))?;
45    let local = listener.local_addr()?;
46    let base = ap::base_url(configured, local);
47    let names = vec![username.clone()];
48    let instance = Arc::new(
49        tokio::task::spawn_blocking(move || Instance::new(&base, &names))
50            .await
51            .context("key generation stopped")??,
52    );
53    let actor_id = instance.actor_url(&username);
54    Log::new(Some(&ctx.status_tx)).info(format!(
55        "ActivityPub client: actor {} ({actor_id}) served on {local}",
56        instance.handle(&username)
57    ));
58
59    let (inbound_tx, mut inbound_rx) = mpsc::channel(ap::INBOX_QUEUE);
60    let endpoint_instance = instance.clone();
61    let endpoint_log = ctx.status_tx.clone();
62    let endpoint_state = ctx.state.clone();
63    let client_id = ctx.client_id;
64    let endpoint = tokio::spawn(async move {
65        let permits = Arc::new(tokio::sync::Semaphore::new(MAX_ENDPOINT_CONNECTIONS));
66        loop {
67            let Ok((stream, _)) = listener.accept().await else {
68                break;
69            };
70            let Ok(permit) = permits.clone().try_acquire_owned() else {
71                continue;
72            };
73            let instance = endpoint_instance.clone();
74            let tx = inbound_tx.clone();
75            let log = endpoint_log.clone();
76            let conn = async move {
77                let _permit = permit;
78                let service = service_fn(move |request| {
79                    let instance = instance.clone();
80                    let tx = tx.clone();
81                    let log = log.clone();
82                    async move {
83                        let line = format!("{} {}", request.method(), request.uri().path());
84                        let (reply, outcome) = ap::http::handle(&instance, request, &tx).await;
85                        if let ap::http::Outcome::Refused(why) = outcome {
86                            Log::new(Some(&log))
87                                .warn(format!("ActivityPub client {line} refused: {why}"));
88                        }
89                        Ok::<_, Infallible>(reply)
90                    }
91                });
92                let _ = http1::Builder::new()
93                    .timer(TokioTimer::new())
94                    .header_read_timeout(ap::HEADER_TIMEOUT)
95                    .serve_connection(TokioIo::new(stream), service)
96                    .await;
97            };
98            let handle = tokio::spawn(conn);
99            endpoint_state.register_client_task(client_id, handle).await;
100        }
101    });
102    ctx.state
103        .register_client_task(ctx.client_id, endpoint)
104        .await;
105
106    ctx.state
107        .update_client_status(ctx.client_id, ClientStatus::Connected)
108        .await;
109    let external =
110        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
111    let (internal_tx, internal_rx) = mpsc::channel::<(Value, usize)>(32);
112    let (event_tx, mut event_rx) = mpsc::channel::<(Event, usize)>(64);
113    event_tx.try_send((
114        Event::new(
115            &actions::READY_EVENT,
116            json!({"actor_id": actor_id, "handle": instance.handle(&username), "remote": ctx.remote_addr}),
117        ),
118        0,
119    ))?;
120
121    // Verified inbox activities become events, after Rust's own bookkeeping.
122    let inbox_instance = instance.clone();
123    let inbox_events = event_tx.clone();
124    let inbox = tokio::spawn(async move {
125        while let Some(inb) = inbound_rx.recv().await {
126            ap::bookkeeping(&inbox_instance, &inb);
127            let mut data = ap::event_data(&inbox_instance, &inb).await;
128            if let Some(o) = data.as_object_mut() {
129                o.remove("to_actor");
130            }
131            if inbox_events
132                .send((Event::new(&actions::ACTIVITY_EVENT, data), 0))
133                .await
134                .is_err()
135            {
136                break;
137            }
138        }
139    });
140    ctx.state.register_client_task(ctx.client_id, inbox).await;
141
142    let events_ctx = ctx.clone();
143    let dispatcher = tokio::spawn(async move {
144        while let Some((event, depth)) = event_rx.recv().await {
145            events_ctx
146                .state
147                .record_access_log(
148                    AccessLogOwner::Client(events_ctx.client_id.as_u32()),
149                    "ActivityPub",
150                    None,
151                    event.id(),
152                    event.data.clone(),
153                    vec![],
154                )
155                .await;
156            let instruction = events_ctx
157                .state
158                .get_instruction_for_client(events_ctx.client_id)
159                .await
160                .unwrap_or_default();
161            let memory = events_ctx
162                .state
163                .get_memory_for_client(events_ctx.client_id)
164                .await
165                .unwrap_or_default();
166            match call_llm_for_client(
167                &events_ctx.llm_client,
168                &events_ctx.state,
169                events_ctx.client_id.to_string(),
170                &instruction,
171                &memory,
172                Some(&event),
173                &ActivityPubClientProtocol,
174                &events_ctx.status_tx,
175            )
176            .await
177            {
178                Ok(result) => {
179                    if let Some(memory) = result.memory_updates {
180                        events_ctx
181                            .state
182                            .set_memory_for_client(events_ctx.client_id, memory)
183                            .await;
184                    }
185                    for action in result.actions {
186                        if internal_tx.send((action, depth + 1)).await.is_err() {
187                            return;
188                        }
189                    }
190                }
191                Err(e) => Log::new(Some(&events_ctx.status_tx))
192                    .warn(format!("ActivityPub client handler: {e}")),
193            }
194        }
195    });
196    let dispatcher_abort = dispatcher.abort_handle();
197    ctx.state
198        .register_client_task(ctx.client_id, dispatcher)
199        .await;
200
201    let session_ctx = ctx.clone();
202    let task = tokio::spawn(async move {
203        session(
204            &session_ctx,
205            &instance,
206            &username,
207            external,
208            internal_rx,
209            event_tx,
210        )
211        .await;
212        dispatcher_abort.abort();
213        session_ctx
214            .state
215            .update_client_status(session_ctx.client_id, ClientStatus::Disconnected)
216            .await;
217        session_ctx
218            .state
219            .remove_client_handle(session_ctx.client_id)
220            .await;
221        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
222    });
223    ctx.state.register_client_task(ctx.client_id, task).await;
224    Ok(local)
225}
226
227async fn perform(instance: &Instance, me: &str, action: &Value) -> Result<Value> {
228    match action["type"].as_str().unwrap_or_default() {
229        actions::LOOKUP => {
230            let id = instance
231                .resolve(action["target"].as_str().unwrap_or_default())
232                .await?;
233            let doc = instance.fetch(&id, Some(me)).await?;
234            Ok(
235                json!({"id": doc["id"], "type": doc["type"], "preferred_username": doc["preferredUsername"],
236                      "name": doc["name"], "summary": doc["summary"].as_str().map(ap::instance::plain_text),
237                      "inbox": doc["inbox"], "followers": doc["followers"],
238                      "manually_approves_followers": doc["manuallyApprovesFollowers"]}),
239            )
240        }
241        actions::FETCH => {
242            instance
243                .fetch(action["url"].as_str().unwrap_or_default(), Some(me))
244                .await
245        }
246        _ => ap::apply_shared(instance, me, action, None).await,
247    }
248}
249
250async fn session(
251    ctx: &ConnectContext,
252    instance: &Instance,
253    me: &str,
254    mut external: mpsc::Receiver<ClientCommand>,
255    mut internal: mpsc::Receiver<(Value, usize)>,
256    events: mpsc::Sender<(Event, usize)>,
257) {
258    let log = Log::new(Some(&ctx.status_tx));
259    loop {
260        let (action, depth, mut injected) = tokio::select! {
261            command = external.recv() => match command {
262                Some(c) => (c.action.clone(), 0, Some(c)),
263                None => return,
264            },
265            action = internal.recv() => match action {
266                Some((a, depth)) => (a, depth, None),
267                None => return,
268            },
269        };
270        let reply = |injected: &mut Option<ClientCommand>, outcome: ClientSendOutcome| {
271            if let Some(command) = injected.take() {
272                crate::client::command_support::reply(command, Ok(outcome));
273            }
274        };
275        match ActivityPubClientProtocol.execute_action(action.clone()) {
276            Ok(ClientActionResult::Disconnect) => {
277                reply(&mut injected, ClientSendOutcome::Disconnected);
278                return;
279            }
280            Ok(_) => {}
281            Err(e) => {
282                log.warn(format!("ActivityPub client action refused: {e:#}"));
283                reply(
284                    &mut injected,
285                    ClientSendOutcome::Rejected {
286                        error: e.to_string(),
287                    },
288                );
289                continue;
290            }
291        }
292        if depth > MAX_FOLLOWUP_DEPTH {
293            log.warn(format!(
294                "ActivityPub client: handler chain stopped after {MAX_FOLLOWUP_DEPTH} follow-ups"
295            ));
296            continue;
297        }
298        if injected.is_some() {
299            ctx.state
300                .record_access_log(
301                    AccessLogOwner::Client(ctx.client_id.as_u32()),
302                    "ActivityPub",
303                    None,
304                    "injected_action",
305                    action.clone(),
306                    vec![],
307                )
308                .await;
309        }
310        let operation = action["type"].as_str().unwrap_or_default();
311        let data = match perform(instance, me, &action).await {
312            Ok(result) => json!({"operation": operation, "ok": true, "result": result}),
313            Err(e) => json!({"operation": operation, "ok": false, "error": format!("{e:#}")}),
314        };
315        reply(
316            &mut injected,
317            ClientSendOutcome::Executed {
318                detail: data.to_string(),
319            },
320        );
321        if events
322            .try_send((Event::new(&actions::RESPONSE_EVENT, data), depth))
323            .is_err()
324        {
325            log.warn("ActivityPub client: event queue full; consumer stalled".to_string());
326            return;
327        }
328    }
329}
netget/client/activitypub/mod.rs

netget/client/activitypub/
mod.rs
