ambee/giterated

1

use anyhow::Error;

2

use connection::{connection_worker, Connections, RawConnection};

3

use giterated_daemon::{

4

authentication::AuthenticationTokenGranter,

5

backend::{

6

discovery::GiteratedDiscoveryProtocol, git::GitBackend, user::UserAuth, DiscoveryBackend,

7

RepositoryBackend, UserBackend,

8

},

9

connection, listener,

10

model::instance::Instance,

11

};

12

use listener::Listeners;

13

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

14

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

15

use tokio::{

16

fs::File,

17

io::{AsyncRead, AsyncReadExt, AsyncWrite},

18

net::{TcpListener, TcpStream},

19

sync::Mutex,

20

};

21

use tokio_tungstenite::{accept_async, WebSocketStream};

use toml::Table;

#[macro_use]

extern crate tracing;

26

27

#[tokio::main]

28

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

29

tracing_subscriber::fmt::init();

30

let mut listener = TcpListener::bind("0.0.0.0:7270").await?;

31

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

32

let listeners: Arc<Mutex<Listeners>> = Arc::default();

33

let config: Table = {

34

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

35

let mut text = String::new();

36

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

37

text.parse()?

38

};

39

let db_conn_options = PgConnectOptions::new()

40

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

41

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

42

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

43

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

44

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

45

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

46

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

47

48

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

49

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

50

info!("Connected");

51

52

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

53

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

54

pg_pool: db_pool.clone(),

55

repository_folder: String::from(

56

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

.as_str()

.unwrap(),

),

}));

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

63

config: config.clone(),

64

}));

65

66

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

67

db_pool,

68

&Instance::from_str("giterated.dev").unwrap(),

69

token_granter.clone(),

70

)));

71

72

let discovery_backend: Arc<Mutex<dyn DiscoveryBackend + Send>> =

73

Arc::new(Mutex::new(GiteratedDiscoveryProtocol {}));

info!("Connected");

loop {

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

79

info!("Connected");

80

81

let (stream, address) = match stream {

82

Ok(stream) => stream,

83

Err(err) => {

84

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

continue;

}

};

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

90

91

let connection = accept_websocket_connection(stream).await;

92

93

let connection = match connection {

94

Ok(connection) => connection,

95

Err(err) => {

96

error!(

97

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

address, err

);

continue;

}

};

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

105

106

let connection = RawConnection {

107

task: tokio::spawn(connection_worker(

connection,

listeners.clone(),

connections.clone(),

repository_backend.clone(),

112

user_backend.clone(),

113

token_granter.clone(),

114

discovery_backend.clone(),

address,

)),

};

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

}

}

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

124

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

Ok(stream)

}

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

130

stream: S,

131

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

132

let connection = accept_async(stream).await?;

Ok(connection)

}