netget/server/activitypub/
mod.rs

1//! ActivityPub instance (server role). HTTP is served by `http.rs` over `instance.rs`;
2//! every activity that arrives at an inbox with a valid signature is queued and answered by
3//! the model one at a time, in order.
4pub mod actions;
5pub mod http;
6pub mod instance;
7pub mod sig;
8
9use crate::llm::action_helper::call_llm;
10use crate::llm::actions::protocol_trait::ActionResult;
11use crate::logging::emit::Log;
12use crate::protocol::{Event, SpawnContext};
13use crate::server::{
14    accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS},
15    connection::ConnectionId,
16};
17use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
18use anyhow::{Context, Result};
19use hyper::{server::conn::http1, service::service_fn};
20use hyper_util::rt::{TokioIo, TokioTimer};
21use instance::{Inbound, Instance};
22use serde_json::{json, Value};
23use std::convert::Infallible;
24use std::net::SocketAddr;
25use std::sync::Arc;
26use std::time::Duration;
27use tokio::sync::mpsc;
28
29pub const HEADER_TIMEOUT: Duration = Duration::from_secs(30);
30/// Verified activities waiting for the model; past it an inbox answers 503.
31pub const INBOX_QUEUE: usize = 256;
32
33/// The base URL ids are built from: the operator's, or http:// and the bound address.
34pub fn base_url(configured: Option<String>, local: SocketAddr) -> String {
35    configured.unwrap_or_else(|| {
36        let host = if local.ip().is_unspecified() {
37            "127.0.0.1".to_string()
38        } else {
39            match local.ip() {
40                std::net::IpAddr::V6(v6) => format!("[{v6}]"),
41                ip => ip.to_string(),
42            }
43        };
44        format!("http://{host}:{}", local.port())
45    })
46}
47
48pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
49    let params = ctx.startup_params.as_ref();
50    let names: Vec<String> = match params
51        .map(|p| p.get_optional_array("actors"))
52        .transpose()?
53        .flatten()
54    {
55        Some(list) => list
56            .iter()
57            .map(|v| {
58                v.as_str()
59                    .map(str::to_string)
60                    .context("actors must be usernames")
61            })
62            .collect::<Result<_>>()?,
63        None => actions::DEFAULT_ACTORS
64            .iter()
65            .map(|s| s.to_string())
66            .collect(),
67    };
68    let configured = params
69        .map(|p| p.get_optional_string("base_url"))
70        .transpose()?
71        .flatten();
72    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
73    let local = listener.local_addr()?;
74    let base = base_url(configured, local);
75    let instance = Arc::new(
76        tokio::task::spawn_blocking(move || Instance::new(&base, &names))
77            .await
78            .context("key generation stopped")??,
79    );
80    Log::new(Some(&ctx.status_tx)).info(format!(
81        "ActivityPub instance {} on {local}, actors: {}",
82        instance.base,
83        instance
84            .actors
85            .lock()
86            .unwrap_or_else(|e| e.into_inner())
87            .keys()
88            .map(|n| instance.handle(n))
89            .collect::<Vec<_>>()
90            .join(", ")
91    ));
92    let (tx, rx) = mpsc::channel(INBOX_QUEUE);
93    let server_id = ctx.server_id;
94    let consumer_ctx = ctx.clone();
95    let consumer_instance = instance.clone();
96    ctx.state
97        .spawn_server_task(server_id, consume(consumer_ctx, consumer_instance, rx))
98        .await;
99    let state = ctx.state.clone();
100    let accept = tokio::spawn(async move {
101        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
102        loop {
103            let (stream, peer, permit) =
104                match accept_bounded(&listener, &limiter, b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", "ActivityPub", Some(&ctx.status_tx)).await {
105                    Ok(v) => v,
106                    Err(_) => break,
107                };
108            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
109            let now = crate::utils::clock::Instant::now();
110            ctx.state
111                .add_connection_to_server(
112                    server_id,
113                    ConnectionState {
114                        id,
115                        remote_addr: peer,
116                        local_addr: local,
117                        bytes_sent: 0,
118                        bytes_received: 0,
119                        packets_sent: 0,
120                        packets_received: 0,
121                        last_activity: now,
122                        status: ConnectionStatus::Active,
123                        status_changed_at: now,
124                        protocol_info: ProtocolConnectionInfo::empty(),
125                    },
126                )
127                .await;
128            let child = ctx.clone();
129            let instance = instance.clone();
130            let tx = tx.clone();
131            ctx.state
132                .spawn_server_task(server_id, async move {
133                    let _permit = permit;
134                    let log_ctx = child.clone();
135                    let service = service_fn(move |request| {
136                        let instance = instance.clone();
137                        let tx = tx.clone();
138                        let ctx = log_ctx.clone();
139                        async move {
140                            let line = format!("{} {}", request.method(), request.uri().path());
141                            let (reply, outcome) = http::handle(&instance, request, &tx).await;
142                            let log = Log::new(Some(&ctx.status_tx));
143                            match outcome {
144                                http::Outcome::Refused(why) => {
145                                    log.warn(format!("ActivityPub {line} refused: {why}"))
146                                }
147                                http::Outcome::Queued => {
148                                    log.info(format!("ActivityPub {line} queued for the model"))
149                                }
150                                http::Outcome::Served => {
151                                    log.debug(format!("ActivityPub {line} {}", reply.status()))
152                                }
153                            }
154                            Ok::<_, Infallible>(reply)
155                        }
156                    });
157                    let _ = http1::Builder::new()
158                        .timer(TokioTimer::new())
159                        .header_read_timeout(HEADER_TIMEOUT)
160                        .max_buf_size(64 * 1024)
161                        .serve_connection(TokioIo::new(stream), service)
162                        .await;
163                    child
164                        .state
165                        .update_connection_status(child.server_id, id, ConnectionStatus::Closed)
166                        .await;
167                    let _ = child.status_tx.send("__UPDATE_UI__".into());
168                })
169                .await;
170        }
171    });
172    state.register_server_task(server_id, accept).await;
173    Ok(local)
174}
175
176/// The local actor an action acts as.
177pub fn acting(instance: &Instance, action: &Value, inbox_owner: Option<&str>) -> Result<String> {
178    let name = action["as"]
179        .as_str()
180        .map(str::to_string)
181        .or_else(|| inbox_owner.map(str::to_string))
182        .or_else(|| instance.first_actor())
183        .context("the instance has no actors")?;
184    anyhow::ensure!(instance.has_actor(&name), "no local actor {name}");
185    Ok(name)
186}
187
188/// Carry out a post, like, follow or unfollow; what happened, for the log or the handler.
189pub async fn apply_shared(
190    instance: &Instance,
191    me: &str,
192    action: &Value,
193    sender: Option<&str>,
194) -> Result<Value> {
195    Ok(match action["type"].as_str().unwrap_or_default() {
196        actions::POST => {
197            let to: Vec<String> = action["to"]
198                .as_array()
199                .map(|a| {
200                    a.iter()
201                        .filter_map(|v| v.as_str().map(str::to_string))
202                        .collect()
203                })
204                .unwrap_or_default();
205            let (create, results) = instance
206                .post(
207                    me,
208                    action["content"].as_str().unwrap_or_default(),
209                    &to,
210                    action["public"].as_bool().unwrap_or(true),
211                    action["in_reply_to"].as_str(),
212                )
213                .await?;
214            json!({"note_id": create["object"]["id"], "deliveries": results.iter().map(|(inbox, r)| match r {
215                Ok(s) => json!({"inbox": inbox, "status": s}),
216                Err(e) => json!({"inbox": inbox, "error": e}),
217            }).collect::<Vec<_>>()})
218        }
219        actions::LIKE => {
220            let to = action["to"]
221                .as_str()
222                .or(sender)
223                .context("whom to tell: give to")?;
224            json!({"status": instance.like(me, action["object"].as_str().unwrap_or_default(), to).await?})
225        }
226        actions::FOLLOW => {
227            let (id, status) = instance
228                .follow(me, action["target"].as_str().unwrap_or_default())
229                .await?;
230            json!({"actor": id, "status": status})
231        }
232        actions::UNFOLLOW => {
233            let (id, status) = instance
234                .unfollow(me, action["target"].as_str().unwrap_or_default())
235                .await?;
236            json!({"actor": id, "status": status})
237        }
238        other => anyhow::bail!("{other} is not something an actor does on its own"),
239    })
240}
241
242/// What Rust settles before the model hears of an activity: follower lists and follows.
243pub fn bookkeeping(instance: &Instance, inb: &Inbound) {
244    let a = &inb.activity;
245    let obj = &a["object"];
246    let to = inb.to.clone().or_else(|| {
247        obj["object"]
248            .as_str()
249            .or(obj.as_str())
250            .and_then(|o| instance.local_name(o))
251    });
252    match (a["type"].as_str(), obj["type"].as_str()) {
253        (Some("Undo"), Some("Follow")) => {
254            if let Some(name) = obj["object"].as_str().and_then(|o| instance.local_name(o)) {
255                instance.remove_follower(&name, &inb.signer);
256            }
257        }
258        (Some(t @ ("Accept" | "Reject")), _) => {
259            if let Some(name) = to {
260                instance.follow_answered(&name, &inb.signer, t == "Accept");
261            }
262        }
263        _ => {}
264    }
265}
266
267/// The event the model is shown for an activity.
268pub async fn event_data(instance: &Instance, inb: &Inbound) -> Value {
269    let mut data = instance::summary(&inb.activity);
270    data["to_actor"] = json!(inb.to);
271    data["actor"] = json!(inb.signer);
272    if let Ok(r) = instance.remote_actor(&inb.signer, inb.to.as_deref()).await {
273        if let (Some(user), Ok(u)) = (r.preferred_username, reqwest::Url::parse(&r.id)) {
274            let host = match u.port() {
275                Some(p) => format!("{}:{p}", u.host_str().unwrap_or_default()),
276                None => u.host_str().unwrap_or_default().to_string(),
277            };
278            data["actor_handle"] = json!(format!("{user}@{host}"));
279        }
280    }
281    data
282}
283
284async fn consume(ctx: SpawnContext, instance: Arc<Instance>, mut rx: mpsc::Receiver<Inbound>) {
285    let log = Log::new(Some(&ctx.status_tx));
286    while let Some(inb) = rx.recv().await {
287        bookkeeping(&instance, &inb);
288        let data = event_data(&instance, &inb).await;
289        let kind = data["type"].as_str().unwrap_or("?").to_string();
290        let event = Event::new(&actions::ACTIVITY_EVENT, data);
291        let result = match call_llm(
292            &ctx.llm_client,
293            &ctx.state,
294            ctx.server_id,
295            None,
296            &event,
297            &actions::ActivityPubProtocol,
298        )
299        .await
300        {
301            Ok(r) if r.failures.is_empty() => r,
302            _ => {
303                log.error(format!(
304                    "ActivityPub {kind} from {} decision=fail_closed_llm_error (nothing sent)",
305                    inb.signer
306                ));
307                continue;
308            }
309        };
310        let mut todo = Vec::new();
311        let mut stack = result.protocol_results;
312        while let Some(r) = stack.pop() {
313            match r {
314                ActionResult::Custom { data, .. } => todo.push(data),
315                ActionResult::Multiple(items) => stack.extend(items),
316                _ => {}
317            }
318        }
319        todo.reverse();
320        log.info(format!(
321            "ActivityPub {kind} from {} decision={}",
322            inb.signer,
323            if todo.is_empty() {
324                "model_silent"
325            } else {
326                "model_answer"
327            }
328        ));
329        for action in todo {
330            let outcome: Result<Value> = async {
331                let me = acting(&instance, &action, inb.to.as_deref())?;
332                match action["type"].as_str().unwrap_or_default() {
333                    t @ (actions::ACCEPT | actions::REJECT) => {
334                        anyhow::ensure!(kind == "Follow", "{t} answers a Follow, not a {kind}");
335                        let me = inb.activity["object"]
336                            .as_str()
337                            .and_then(|o| instance.local_name(o))
338                            .unwrap_or(me);
339                        let status = instance
340                            .answer_follow(&me, &inb.activity, t == actions::ACCEPT)
341                            .await?;
342                        Ok(json!({"status": status}))
343                    }
344                    _ => apply_shared(&instance, &me, &action, Some(&inb.signer)).await,
345                }
346            }
347            .await;
348            match outcome {
349                Ok(v) => log.info(format!("ActivityPub {}: {v}", action["type"])),
350                Err(e) => log.warn(format!("ActivityPub {} failed: {e:#}", action["type"])),
351            }
352        }
353    }
354}
netget/server/activitypub/mod.rs

netget/server/activitypub/
mod.rs
