orign 0.2.3

A globally distributed container orchestrator
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

// src/lib.rs

pub mod auth;
pub mod buffers;
pub mod config;
pub mod db;
pub mod entities;
pub mod errors;
pub mod handlers;
pub mod humans;
pub mod middleware;
pub mod models;
pub mod mutation;
pub mod org;
pub mod query;
pub mod resources;
pub mod routes;
pub mod state;
pub mod validate;

use crate::config::CONFIG;
use axum::Router;
use db::init_db;
use rdkafka::admin::AdminClient;
use rdkafka::producer::FutureProducer;
use rdkafka::ClientConfig;
use routes::create_routes;
use state::AppState;
use state::MessageQueue;
use std::env;
use std::sync::Arc;
use tower_http::cors::{Any, CorsLayer};
use tower_http::trace::TraceLayer;
use tracing::debug;
use url::Url;

pub async fn create_app() -> Result<Router, Box<dyn std::error::Error>> {
    println!("Creating app");
    let db_pool = init_db().await?;
    println!("Database pool created");

    // Initialize the appropriate message queue based on configuration
    let message_queue = match CONFIG.message_queue_type.to_lowercase().as_str() {
        "redis" => {
            let redis_url = if let Some(url) = &CONFIG.redis_url {
                if !url.is_empty() {
                    // Redis URL exists, so use it directly but also parse it to set env vars
                    if let Ok(parsed_url) = Url::parse(url) {
                        // Extract and set host
                        if let Some(host) = parsed_url.host_str() {
                            env::set_var("REDIS_HOST", host);
                        }

                        // Extract and set port
                        if let Some(port) = parsed_url.port() {
                            env::set_var("REDIS_PORT", port.to_string());
                        } else {
                            // Default redis port if not specified in URL
                            env::set_var("REDIS_PORT", "6379");
                        }

                        // Extract and set password if present
                        if let Some(password) = parsed_url.password() {
                            env::set_var("REDIS_PASSWORD", password);
                        }
                    }

                    url.clone()
                } else {
                    build_redis_url_from_components()
                }
            } else {
                build_redis_url_from_components()
            };

            debug!("Redis URL: {}", redis_url);

            // Create the Redis client using the constructed URL
            let redis_client = Arc::new(redis::Client::open(redis_url.as_str())?);

            MessageQueue::Redis {
                client: redis_client,
            }
        }
        "kafka" => {
            let mut client_config = ClientConfig::new();
            let kafka_config = client_config
                .set("bootstrap.servers", &CONFIG.kafka_bootstrap_servers)
                .set("message.timeout.ms", &CONFIG.kafka_timeout_ms);

            let producer = Arc::new(kafka_config.clone().create::<FutureProducer>()?);
            let admin = Arc::new(kafka_config.create::<AdminClient<_>>()?);

            MessageQueue::Kafka { producer, admin }
        }
        unsupported => {
            return Err(format!("Unsupported message queue type: {}", unsupported).into())
        }
    };

    let app_state = AppState {
        db_pool: db_pool.clone(),
        message_queue,
    };

    // Create your routes
    let routes = create_routes(app_state.clone());

    // Define a CORS layer (this example allows any origin, headers, and methods)
    let cors = CorsLayer::new()
        .allow_origin(Any) // TODO: fix me
        .allow_methods(Any)
        .allow_headers(Any);

    let app = routes
        .layer(TraceLayer::new_for_http())
        .layer(cors)
        .with_state(app_state);

    Ok(app)
}

fn build_redis_url_from_components() -> String {
    // Redis URL not present or empty, build from components
    let host = &CONFIG.redis_host;
    let port = &CONFIG.redis_port;

    match &CONFIG.redis_password {
        Some(password) if !password.is_empty() => {
            format!("redis://:{}@{}:{}", password, host, port)
        }
        _ => format!("redis://{}:{}", host, port),
    }
}