1use std::{fmt, sync::Arc};
2
3use futures::{StreamExt, TryStreamExt};
4use tokio::sync::Mutex;
5use tuwunel_core::{
6 Result, Server, debug, debug_info, implement, info, trace, utils::stream::IterStream,
7};
8use tuwunel_database::Database;
9
10pub(crate) use crate::OnceServices;
11use crate::{
12 account_data, admin, appservice, client, config, deactivate, emergency, federation, fetcher,
13 globals, key_backups,
14 manager::Manager,
15 media, membership, oauth, presence, profile, pusher, registration_tokens, rendezvous,
16 resolver,
17 rooms::{self, retention},
18 sending, sendmail, server_keys,
19 service::{Args, Service},
20 storage, sync, tasks, threepid, transaction_ids, uiaa, users,
21};
22
23pub struct Services {
24 pub account_data: Arc<account_data::Service>,
25 pub admin: Arc<admin::Service>,
26 pub appservice: Arc<appservice::Service>,
27 pub config: Arc<config::Service>,
28 pub client: Arc<client::Service>,
29 pub emergency: Arc<emergency::Service>,
30 pub fetcher: Arc<fetcher::Service>,
31 pub globals: Arc<globals::Service>,
32 pub key_backups: Arc<key_backups::Service>,
33 pub media: Arc<media::Service>,
34 pub presence: Arc<presence::Service>,
35 pub pusher: Arc<pusher::Service>,
36 pub resolver: Arc<resolver::Service>,
37 pub alias: Arc<rooms::alias::Service>,
38 pub auth_chain: Arc<rooms::auth_chain::Service>,
39 pub delete: Arc<rooms::delete::Service>,
40 pub directory: Arc<rooms::directory::Service>,
41 pub event_handler: Arc<rooms::event_handler::Service>,
42 pub lazy_loading: Arc<rooms::lazy_loading::Service>,
43 pub metadata: Arc<rooms::metadata::Service>,
44 pub pdu_metadata: Arc<rooms::pdu_metadata::Service>,
45 pub read_receipt: Arc<rooms::read_receipt::Service>,
46 pub search: Arc<rooms::search::Service>,
47 pub short: Arc<rooms::short::Service>,
48 pub spaces: Arc<rooms::spaces::Service>,
49 pub state: Arc<rooms::state::Service>,
50 pub state_accessor: Arc<rooms::state_accessor::Service>,
51 pub state_cache: Arc<rooms::state_cache::Service>,
52 pub state_compressor: Arc<rooms::state_compressor::Service>,
53 pub storage: Arc<storage::Service>,
54 pub threads: Arc<rooms::threads::Service>,
55 pub timeline: Arc<rooms::timeline::Service>,
56 pub typing: Arc<rooms::typing::Service>,
57 pub federation: Arc<federation::Service>,
58 pub sending: Arc<sending::Service>,
59 pub server_keys: Arc<server_keys::Service>,
60 pub sync: Arc<sync::Service>,
61 pub tasks: Arc<tasks::Service>,
62 pub transaction_ids: Arc<transaction_ids::Service>,
63 pub uiaa: Arc<uiaa::Service>,
64 pub users: Arc<users::Service>,
65 pub membership: Arc<membership::Service>,
66 pub deactivate: Arc<deactivate::Service>,
67 pub oauth: Arc<oauth::Service>,
68 pub retention: Arc<retention::Service>,
69 pub registration_tokens: Arc<registration_tokens::Service>,
70 pub rendezvous: Arc<rendezvous::Service>,
71 pub sendmail: Arc<sendmail::Service>,
72 pub threepid: Arc<threepid::Service>,
73 pub profile: Arc<profile::Service>,
74
75 manager: Mutex<Option<Arc<Manager>>>,
76 pub server: Arc<Server>,
77 pub db: Arc<Database>,
78}
79
80#[implement(Services)]
81pub async fn build(server: Arc<Server>) -> Result<Arc<Self>> {
82 let db = Database::open(&server).await?;
83 let services = Arc::new(OnceServices::default());
84 let args = Args {
85 db: &db,
86 server: &server,
87 services: &services,
88 };
89
90 let res = Arc::new(Self {
91 account_data: account_data::Service::build(&args)?,
92 admin: admin::Service::build(&args)?,
93 appservice: appservice::Service::build(&args)?,
94 resolver: resolver::Service::build(&args)?,
95 client: client::Service::build(&args)?,
96 config: config::Service::build(&args)?,
97 emergency: emergency::Service::build(&args)?,
98 fetcher: fetcher::Service::build(&args)?,
99 globals: globals::Service::build(&args)?,
100 key_backups: key_backups::Service::build(&args)?,
101 media: media::Service::build(&args)?,
102 presence: presence::Service::build(&args)?,
103 pusher: pusher::Service::build(&args)?,
104 alias: rooms::alias::Service::build(&args)?,
105 auth_chain: rooms::auth_chain::Service::build(&args)?,
106 delete: rooms::delete::Service::build(&args)?,
107 directory: rooms::directory::Service::build(&args)?,
108 event_handler: rooms::event_handler::Service::build(&args)?,
109 lazy_loading: rooms::lazy_loading::Service::build(&args)?,
110 metadata: rooms::metadata::Service::build(&args)?,
111 pdu_metadata: rooms::pdu_metadata::Service::build(&args)?,
112 read_receipt: rooms::read_receipt::Service::build(&args)?,
113 search: rooms::search::Service::build(&args)?,
114 short: rooms::short::Service::build(&args)?,
115 spaces: rooms::spaces::Service::build(&args)?,
116 state: rooms::state::Service::build(&args)?,
117 state_accessor: rooms::state_accessor::Service::build(&args)?,
118 state_cache: rooms::state_cache::Service::build(&args)?,
119 state_compressor: rooms::state_compressor::Service::build(&args)?,
120 storage: storage::Service::build(&args)?,
121 threads: rooms::threads::Service::build(&args)?,
122 timeline: rooms::timeline::Service::build(&args)?,
123 typing: rooms::typing::Service::build(&args)?,
124 federation: federation::Service::build(&args)?,
125 sending: sending::Service::build(&args)?,
126 server_keys: server_keys::Service::build(&args)?,
127 sync: sync::Service::build(&args)?,
128 tasks: tasks::Service::build(&args)?,
129 transaction_ids: transaction_ids::Service::build(&args)?,
130 uiaa: uiaa::Service::build(&args)?,
131 users: users::Service::build(&args)?,
132 membership: membership::Service::build(&args)?,
133 deactivate: deactivate::Service::build(&args)?,
134 oauth: oauth::Service::build(&args)?,
135 retention: retention::Service::build(&args)?,
136 registration_tokens: registration_tokens::Service::build(&args)?,
137 rendezvous: rendezvous::Service::build(&args)?,
138 sendmail: sendmail::Service::build(&args)?,
139 threepid: threepid::Service::build(&args)?,
140 profile: profile::Service::build(&args)?,
141
142 manager: Mutex::new(None),
143 server,
144 db,
145 });
146
147 Ok(services.set(res))
148}
149
150#[implement(Services)]
151pub(crate) fn services(&self) -> impl Iterator<Item = Arc<dyn Service>> + Send {
152 macro_rules! cast {
153 ($s:expr) => {
154 <Arc<dyn Service> as Into<_>>::into($s.clone())
155 };
156 }
157
158 [
159 cast!(self.account_data),
160 cast!(self.admin),
161 cast!(self.appservice),
162 cast!(self.resolver),
163 cast!(self.client),
164 cast!(self.config),
165 cast!(self.emergency),
166 cast!(self.fetcher),
167 cast!(self.globals),
168 cast!(self.key_backups),
169 cast!(self.media),
170 cast!(self.presence),
171 cast!(self.pusher),
172 cast!(self.alias),
173 cast!(self.auth_chain),
174 cast!(self.delete),
175 cast!(self.directory),
176 cast!(self.event_handler),
177 cast!(self.lazy_loading),
178 cast!(self.metadata),
179 cast!(self.pdu_metadata),
180 cast!(self.read_receipt),
181 cast!(self.search),
182 cast!(self.short),
183 cast!(self.spaces),
184 cast!(self.state),
185 cast!(self.state_accessor),
186 cast!(self.state_cache),
187 cast!(self.state_compressor),
188 cast!(self.storage),
189 cast!(self.threads),
190 cast!(self.timeline),
191 cast!(self.typing),
192 cast!(self.federation),
193 cast!(self.sending),
194 cast!(self.server_keys),
195 cast!(self.sync),
196 cast!(self.tasks),
197 cast!(self.transaction_ids),
198 cast!(self.uiaa),
199 cast!(self.users),
200 cast!(self.membership),
201 cast!(self.deactivate),
202 cast!(self.oauth),
203 cast!(self.retention),
204 cast!(self.registration_tokens),
205 cast!(self.rendezvous),
206 cast!(self.profile),
207 ]
208 .into_iter()
209}
210
211impl fmt::Debug for Services {
212 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
213 f.debug_struct("Services").finish()
214 }
215}
216
217#[implement(Services)]
218pub async fn start(self: &Arc<Self>) -> Result<Arc<Self>> {
219 debug_info!("Starting services...");
220
221 super::migrations::migrations(self).await?;
222
223 self.manager
224 .lock()
225 .await
226 .insert(Manager::new(self))
227 .clone()
228 .start()
229 .await?;
230
231 debug_info!("Services startup complete.");
232
233 Ok(Arc::clone(self))
234}
235
236#[implement(Services)]
237pub async fn stop(&self) {
238 info!("Shutting down services...");
239
240 self.interrupt().await;
241 if let Some(manager) = self.manager.lock().await.as_ref() {
242 manager.stop().await;
243 }
244
245 debug_info!("Services shutdown complete.");
246}
247
248#[implement(Services)]
249pub(crate) async fn interrupt(&self) {
250 debug!("Interrupting services...");
251 for service in self.services() {
252 trace!(
253 name = ?service.name(),
254 "Interrupting Service"
255 );
256
257 service.interrupt().await;
258 }
259}
260
261#[implement(Services)]
262pub async fn poll(&self) -> Result {
263 if let Some(manager) = self.manager.lock().await.as_ref() {
264 trace!("Polling service manager...");
265 return manager.poll().await;
266 }
267
268 Ok(())
269}
270
271#[implement(Services)]
272pub async fn clear_cache(&self) {
273 let _cork = self.db.cork_and_flush();
276
277 self.services()
278 .stream()
279 .for_each(async |service| {
280 service.clear_cache().await;
281 })
282 .await;
283}
284
285#[implement(Services)]
286pub async fn memory_usage(&self) -> Result<String> {
287 self.services()
288 .try_stream()
289 .try_fold(String::new(), async |mut out, service| {
290 service.memory_usage(&mut out).await?;
291 Ok(out)
292 })
293 .await
294}