telexide_fork/client/
webhook_handling.rs

1use std::{
2    convert::Infallible,
3    io::Write,
4    net::{IpAddr, SocketAddr},
5};
6
7use crate::{
8    model::Update,
9    utils::result::{Result as TelegramResult, TelegramError},
10};
11use hyper::{
12    body::HttpBody,
13    service::{make_service_fn, service_fn},
14    Body, Method, Request, Response, Server, StatusCode, Uri,
15};
16use tokio::sync::mpsc::{channel, Receiver, Sender};
17
18/// Handles listening to the telegram webhook and will provide you with the
19/// incoming updates
20#[derive(Debug)]
21pub struct Webhook {
22    opts: WebhookOptions,
23}
24
25impl Webhook {
26    /// creates a new `Webhook` based on the provided `WebhookOptions`
27    pub fn new(opts: &WebhookOptions) -> Self {
28        Self { opts: opts.clone() }
29    }
30
31    /// starts the webhandling and returns a [`Receiver`], which will allow you
32    /// to receive the incoming updates
33    pub fn start(self) -> Receiver<TelegramResult<Update>> {
34        let (tx, rx) = channel(1000);
35
36        tokio::spawn(start_ws(self.opts, tx));
37        rx
38    }
39}
40
41async fn handle_update(
42    payload: HandlingPayload,
43    req: Request<Body>,
44) -> TelegramResult<Response<Body>> {
45    let mut response = Response::new(Body::empty());
46
47    let mut raw_body = req.into_body();
48    let mut body: Vec<u8> = Vec::new();
49    while let Some(chunk) = raw_body.data().await {
50        body.write_all(&chunk?)?;
51    }
52
53    let update: Update = serde_json::from_slice(&body)?;
54    let send_res = payload.chan.send(Ok(update)).await;
55    if send_res.is_err() {
56        return Err(TelegramError::WebhookError.into());
57    }
58
59    *response.status_mut() = StatusCode::OK;
60    Ok(response)
61}
62
63async fn handle_req(
64    payload: HandlingPayload,
65    req: Request<Body>,
66) -> Result<Response<Body>, Infallible> {
67    let mut response = Response::new(Body::empty());
68
69    match (req.method(), req.uri().path()) {
70        (&Method::POST, path) if path == payload.path => {
71            let result = handle_update(payload, req).await;
72
73            if result.is_err() {
74                *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
75            } else {
76                response = result.unwrap();
77            }
78        }
79        _ => {
80            *response.status_mut() = StatusCode::NOT_FOUND;
81        }
82    }
83
84    Ok(response)
85}
86
87async fn start_ws(
88    opts: WebhookOptions,
89    chan: Sender<TelegramResult<Update>>,
90) -> TelegramResult<()> {
91    let addr = SocketAddr::from((opts.ip, opts.port));
92
93    let payload = HandlingPayload::new(&opts, chan.clone());
94    let make_svc = make_service_fn(move |_conn| {
95        let inner_payload = payload.clone();
96        async move {
97            Ok::<_, Infallible>(service_fn(move |req| {
98                handle_req(inner_payload.clone(), req)
99            }))
100        }
101    });
102
103    let server = Server::bind(&addr).serve(make_svc);
104    let graceful = server.with_graceful_shutdown(shutdown_signal());
105
106    if let Err(e) = graceful.await {
107        let send_res = chan
108            .send(Err(TelegramError::Unknown(e.to_string()).into()))
109            .await;
110        if send_res.is_err() {
111            return Err(TelegramError::WebhookError.into());
112        }
113    }
114    Ok(())
115}
116
117async fn shutdown_signal() {
118    // Wait for the CTRL+C signal
119    tokio::signal::ctrl_c()
120        .await
121        .expect("failed to install CTRL+C signal handler");
122}
123
124/// Represents the options to set for the webhook handling
125#[derive(Clone, Debug)]
126pub struct WebhookOptions {
127    pub url: Option<Uri>,
128    pub path: String,
129    pub port: u16,
130    pub ip: IpAddr,
131}
132
133impl WebhookOptions {
134    /// Creates a new `WebhookOptions` with default values
135    ///
136    /// By default it will listen on 127.0.0.1:8006 and the path being the root
137    pub fn new() -> Self {
138        Self {
139            url: None,
140            path: "/".to_owned(),
141            port: 8006,
142            ip: [127, 0, 0, 1].into(),
143        }
144    }
145
146    /// Sets the path of the webhook
147    pub fn set_path(&mut self, path: &str) -> &mut Self {
148        self.path = path.to_owned();
149        self
150    }
151
152    /// Sets the port the webhook will be listening on
153    pub fn set_port(&mut self, port: u16) -> &mut Self {
154        self.port = port;
155        self
156    }
157
158    /// Sets the IP the webhook will be listening on
159    pub fn set_ip<T: Into<IpAddr>>(&mut self, ip: T) -> &mut Self {
160        self.ip = ip.into();
161        self
162    }
163
164    /// Sets the url of the webhook
165    pub fn set_url(&mut self, url: &str) -> TelegramResult<&mut Self> {
166        self.url = Some(url.parse()?);
167        Ok(self)
168    }
169
170    fn get_path(&self) -> &str {
171        self.url
172            .as_ref()
173            .map_or_else(|| self.path.as_str(), |url| url.path())
174    }
175}
176
177impl Default for WebhookOptions {
178    fn default() -> Self {
179        Self::new()
180    }
181}
182
183#[derive(Clone, Debug)]
184struct HandlingPayload {
185    path: String,
186    chan: Sender<TelegramResult<Update>>,
187}
188
189impl HandlingPayload {
190    fn new(opts: &WebhookOptions, sender: Sender<TelegramResult<Update>>) -> Self {
191        Self {
192            path: opts.get_path().to_owned(),
193            chan: sender,
194        }
195    }
196}
telexide_fork/client/webhook_handling.rs

telexide_fork/client/
webhook_handling.rs
