ambee/giterated

1

use anyhow::Error;

2

use connection::{Connections, RawConnection};

3

use giterated_daemon::{

4

authentication::AuthenticationTokenGranter,

5

backend::{

6

git::GitBackend, settings::DatabaseSettings, user::UserAuth, RepositoryBackend, UserBackend,

7

},

8

connection::{self, wrapper::connection_wrapper},

9

database_backend::DatabaseBackend,

10

federation::connections::InstanceConnections,

11

};

12

13

use giterated_models::instance::Instance;

14

15

use giterated_stack::{GiteratedStack, StackOperationState};

16

use sqlx::{postgres::PgConnectOptions, ConnectOptions, PgPool};

17

use std::{net::SocketAddr, str::FromStr, sync::Arc};

18

use tokio::{

19

fs::File,

20

io::{AsyncRead, AsyncReadExt, AsyncWrite},

21

net::{TcpListener, TcpStream},

22

sync::{Mutex, OnceCell},

23

};

24

use tokio_tungstenite::{accept_async, WebSocketStream};

25

use tokio_util::task::LocalPoolHandle;

use toml::Table;

#[macro_use]

extern crate tracing;

30

31

#[tokio::main]

32

async fn main() -> Result<(), Error> {

33

tracing_subscriber::fmt::init();

34

let config: Table = {

35

let mut file = File::open("Giterated.toml").await?;

36

let mut text = String::new();

37

file.read_to_string(&mut text).await?;

38

text.parse()?

39

};

40

let mut listener = TcpListener::bind(config["giterated"]["bind"].as_str().unwrap()).await?;

41

let connections: Arc<Mutex<Connections>> = Arc::default();

42

let instance_connections: Arc<Mutex<InstanceConnections>> = Arc::default();

43

let db_conn_options = PgConnectOptions::new()

44

.host(config["postgres"]["host"].as_str().unwrap())

45

.port(config["postgres"]["port"].as_integer().unwrap() as u16)

46

.database(config["postgres"]["database"].as_str().unwrap())

47

.username(config["postgres"]["user"].as_str().unwrap())

48

.password(config["postgres"]["password"].as_str().unwrap())

49

.log_statements(log::LevelFilter::Off);

50

let db_pool = PgPool::connect_with(db_conn_options).await?;

51

52

debug!("Running database migrations...");

53

sqlx::migrate!().run(&db_pool).await?;

54

info!("Connected");

55

56

let stack_cell = Arc::new(OnceCell::default());

57

58

let settings = Arc::new(Mutex::new(DatabaseSettings {

59

pg_pool: db_pool.clone(),

60

stack: stack_cell.clone(),

61

}));

62

63

let repository_backend: Arc<Mutex<dyn RepositoryBackend + Send>> =

64

Arc::new(Mutex::new(GitBackend {

65

pg_pool: db_pool.clone(),

66

repository_folder: String::from(

67

config["giterated"]["backend"]["git"]["root"]

.as_str()

.unwrap(),

),

instance: Instance::from_str(config["giterated"]["instance"].as_str().unwrap())

72

.unwrap(),

73

stack: stack_cell.clone(),

74

}));

75

76

let token_granter = Arc::new(Mutex::new(AuthenticationTokenGranter {

77

config: config.clone(),

78

instance: Instance::from_str(config["giterated"]["instance"].as_str().unwrap()).unwrap(),

79

}));

80

81

let user_backend: Arc<Mutex<dyn UserBackend + Send>> = Arc::new(Mutex::new(UserAuth::new(

82

db_pool.clone(),

83

&Instance::from_str(config["giterated"]["instance"].as_str().unwrap()).unwrap(),

84

token_granter.clone(),

settings.clone(),

)));

info!("Connected");

let database_backend = DatabaseBackend::new(

91

Instance::from_str(config["giterated"]["instance"].as_str().unwrap()).unwrap(),

92

user_backend.clone(),

93

repository_backend.clone(),

db_pool.clone(),

stack_cell.clone(),

);

let mut runtime = GiteratedStack::default();

99

100

let database_backend = database_backend.into_substack();

101

runtime.merge_builder(database_backend);

102

103

let runtime = Arc::new(runtime);

104

105

stack_cell

106

.set(runtime.clone())

107

.expect("failed to store global daemon stack");

108

109

let operation_state = {

110

StackOperationState {

111

our_instance: Instance::from_str(config["giterated"]["instance"].as_str().unwrap())

112

.unwrap(),

113

runtime: runtime.clone(),

instance: None,

user: None,

}

};

let pool = LocalPoolHandle::new(5);

120

121

loop {

122

let stream = accept_stream(&mut listener).await;

123

info!("Connected");

124

125

let (stream, address) = match stream {

126

Ok(stream) => stream,

127

Err(err) => {

128

error!("Failed to accept connection. {:?}", err);

continue;

}

};

info!("Accepted connection from {}", address);

134

135

let connection = accept_websocket_connection(stream).await;

136

137

let connection = match connection {

138

Ok(connection) => connection,

139

Err(err) => {

140

error!(

141

"Failed to initiate Websocket connection from {}. {:?}",

address, err

);

continue;

}

};

info!("Websocket connection established with {}", address);

149

let connections_cloned = connections.clone();

150

let repository_backend = repository_backend.clone();

151

let user_backend = user_backend.clone();

152

let token_granter = token_granter.clone();

153

let settings = settings.clone();

154

let instance_connections = instance_connections.clone();

155

let config = config.clone();

156

let runtime = runtime.clone();

157

let operation_state = operation_state.clone();

158

159

pool.spawn_pinned(move || {

connection_wrapper(

connection,

connections_cloned,

repository_backend,

user_backend,

token_granter,

settings,

address,

Instance::from_str(config["giterated"]["instance"].as_str().unwrap()).unwrap(),

169

instance_connections,

config,

runtime,

operation_state,

)

});

let connection = RawConnection {

177

task: tokio::spawn(async move { () }),

178

};

179

180

connections.lock().await.connections.push(connection);

}

}

async fn accept_stream(listener: &mut TcpListener) -> Result<(TcpStream, SocketAddr), Error> {

185

let stream = listener.accept().await?;

Ok(stream)

}

async fn accept_websocket_connection<S: AsyncRead + AsyncWrite + Unpin>(

191

stream: S,

192

) -> Result<WebSocketStream<S>, Error> {

193

let connection = accept_async(stream).await?;

Ok(connection)

}