Merge pull request #1085 from cloudflare/avery/edge-dev

* [dev] route authenticated dev requests to the edge (#1098)

* [dev] connect to edge websocket (#1291)

* [dev] live reload for edge dev (#1452)

src/commands/dev/edge/mod.rs

@@ -0,0 +1,69 @@
+mod server;
+mod setup;
+mod watch;
+
+use server::serve;
+use setup::{upload, Session};
+use watch::watch_for_changes;
+
+use crate::commands::dev::{socket, ServerConfig};
+use crate::settings::global_user::GlobalUser;
+use crate::settings::toml::{DeployConfig, Target};
+
+use tokio::runtime::Runtime as TokioRuntime;
+
+use std::sync::{Arc, Mutex};
+use std::thread;
+
+pub fn dev(
+    target: Target,
+    user: GlobalUser,
+    server_config: ServerConfig,
+    deploy_config: DeployConfig,
+    verbose: bool,
+) -> Result<(), failure::Error> {
+    let session = Session::new(&target, &user, &deploy_config)?;
+    let mut target = target;
+
+    let preview_token = upload(
+        &mut target,
+        &deploy_config,
+        &user,
+        session.preview_token.clone(),
+        verbose,
+    )?;
+
+    let preview_token = Arc::new(Mutex::new(preview_token));
+
+    {
+        let preview_token = preview_token.clone();
+        let session_token = session.preview_token.clone();
+
+        thread::spawn(move || {
+            watch_for_changes(
+                target,
+                &deploy_config,
+                &user,
+                Arc::clone(&preview_token),
+                session_token,
+                verbose,
+            )
+        });
+    }
+
+    let mut runtime = TokioRuntime::new()?;
+    runtime.block_on(async {
+        let devtools_listener = tokio::spawn(socket::listen(session.websocket_url));
+        let server = tokio::spawn(serve(
+            server_config,
+            Arc::clone(&preview_token),
+            session.host,
+        ));
+        let res = tokio::try_join!(async { devtools_listener.await? }, async { server.await? });
+
+        match res {
+            Ok(_) => Ok(()),
+            Err(e) => Err(e),
+        }
+    })
+}

src/commands/dev/edge/server.rs

@@ -0,0 +1,101 @@
+use crate::commands::dev::server_config::ServerConfig;
+use crate::commands::dev::utils::get_path_as_str;
+use crate::terminal::emoji;
+
+use std::sync::{Arc, Mutex};
+
+use chrono::prelude::*;
+use hyper::client::{HttpConnector, ResponseFuture};
+use hyper::header::{HeaderName, HeaderValue};
+use hyper::service::{make_service_fn, service_fn};
+use hyper::{Body, Client as HyperClient, Request, Server};
+use hyper_tls::HttpsConnector;
+
+pub(super) async fn serve(
+    server_config: ServerConfig,
+    preview_token: Arc<Mutex<String>>,
+    host: String,
+) -> Result<(), failure::Error> {
+    // set up https client to connect to the preview service
+    let https = HttpsConnector::new();
+    let client = HyperClient::builder().build::<_, Body>(https);
+
+    let listening_address = server_config.listening_address;
+
+    // create a closure that hyper will use later to handle HTTP requests
+    let make_service = make_service_fn(move |_| {
+        let client = client.to_owned();
+        let preview_token = preview_token.to_owned();
+        let host = host.to_owned();
+
+        async move {
+            Ok::<_, failure::Error>(service_fn(move |req| {
+                let client = client.to_owned();
+                let preview_token = preview_token.lock().unwrap().to_owned();
+                let host = host.to_owned();
+                let version = req.version();
+                let (parts, body) = req.into_parts();
+                let req_method = parts.method.to_string();
+                let now: DateTime<Local> = Local::now();
+                let path = get_path_as_str(&parts.uri);
+                async move {
+                    let resp = preview_request(
+                        Request::from_parts(parts, body),
+                        client,
+                        preview_token.to_owned(),
+                        host.clone(),
+                    )
+                    .await?;
+
+                    println!(
+                        "[{}] {} {}{} {:?} {}",
+                        now.format("%Y-%m-%d %H:%M:%S"),
+                        req_method,
+                        host,
+                        path,
+                        version,
+                        resp.status()
+                    );
+                    Ok::<_, failure::Error>(resp)
+                }
+            }))
+        }
+    });
+
+    let server = Server::bind(&listening_address).serve(make_service);
+    println!("{} Listening on http://{}", emoji::EAR, listening_address);
+    if let Err(e) = server.await {
+        eprintln!("server error: {}", e)
+    }
+    Ok(())
+}
+
+fn preview_request(
+    req: Request<Body>,
+    client: HyperClient<HttpsConnector<HttpConnector>>,
+    preview_token: String,
+    host: String,
+) -> ResponseFuture {
+    let (mut parts, body) = req.into_parts();
+
+    let path = get_path_as_str(&parts.uri);
+
+    parts.headers.insert(
+        HeaderName::from_static("host"),
+        HeaderValue::from_str(&host).expect("Could not create host header"),
+    );
+
+    parts.headers.insert(
+        HeaderName::from_static("cf-workers-preview-token"),
+        HeaderValue::from_str(&preview_token).expect("Could not create token header"),
+    );
+
+    // TODO: figure out how to http _or_ https
+    parts.uri = format!("https://{}{}", host, path)
+        .parse()
+        .expect("Could not construct preview url");
+
+    let req = Request::from_parts(parts, body);
+
+    client.request(req)
+}

src/commands/dev/edge/setup.rs

@@ -0,0 +1,190 @@
+use std::path::Path;
+
+use crate::kv::bulk;
+use crate::settings::global_user::GlobalUser;
+use crate::settings::toml::{DeployConfig, Target};
+use crate::sites::{add_namespace, sync};
+use crate::terminal::message;
+use crate::upload;
+
+use reqwest::Url;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+pub(super) fn upload(
+    target: &mut Target,
+    deploy_config: &DeployConfig,
+    user: &GlobalUser,
+    session_token: String,
+    verbose: bool,
+) -> Result<String, failure::Error> {
+    let client = crate::http::legacy_auth_client(&user);
+
+    let (to_delete, asset_manifest, site_namespace_id) = if let Some(site_config) =
+        target.site.clone()
+    {
+        let site_namespace = add_namespace(user, target, true)?;
+        let path = Path::new(&site_config.bucket);
+        let (to_upload, to_delete, asset_manifest) = sync(target, user, &site_namespace.id, path)?;
+
+        // First, upload all existing files in given directory
+        if verbose {
+            message::info("Uploading updated files...");
+        }
+
+        bulk::put(target, user, &site_namespace.id, to_upload, &None)?;
+        (to_delete, Some(asset_manifest), Some(site_namespace.id))
+    } else {
+        (Vec::new(), None, None)
+    };
+
+    let session_config = get_session_config(deploy_config);
+    let address = get_upload_address(target);
+
+    let script_upload_form = upload::form::build(target, asset_manifest, Some(session_config))?;
+
+    let response = client
+        .post(&address)
+        .header("cf-preview-upload-config-token", session_token)
+        .multipart(script_upload_form)
+        .send()?
+        .error_for_status()?;
+
+    if !to_delete.is_empty() {
+        if verbose {
+            message::info("Deleting stale files...");
+        }
+
+        bulk::delete(target, user, &site_namespace_id.unwrap(), to_delete, &None)?;
+    }
+
+    let text = &response.text()?;
+
+    // TODO: use cloudflare-rs for this :)
+    let response: PreviewV4ApiResponse = serde_json::from_str(text)?;
+    Ok(response.result.preview_token)
+}
+
+#[derive(Debug, Clone)]
+pub struct Session {
+    pub host: String,
+    pub websocket_url: Url,
+    pub preview_token: String,
+}
+
+impl Session {
+    pub fn new(
+        target: &Target,
+        user: &GlobalUser,
+        deploy_config: &DeployConfig,
+    ) -> Result<Session, failure::Error> {
+        let exchange_url = get_exchange_url(deploy_config, user)?;
+        let host = match exchange_url.host_str() {
+            Some(host) => Ok(host.to_string()),
+            None => Err(failure::format_err!(
+                "Could not parse host from exchange url"
+            )),
+        }?;
+
+        let host = match deploy_config {
+            DeployConfig::Zoned(_) => host,
+            DeployConfig::Zoneless(_) => {
+                let namespaces: Vec<&str> = host.as_str().split('.').collect();
+                let subdomain = namespaces[1];
+                format!("{}.{}.workers.dev", target.name, subdomain)
+            }
+        };
+
+        let client = crate::http::legacy_auth_client(&user);
+        let response = client.get(exchange_url).send()?.error_for_status()?;
+        let text = &response.text()?;
+        let response: InspectorV4ApiResponse = serde_json::from_str(text)?;
+        let full_url = format!(
+            "{}?{}={}",
+            &response.inspector_websocket, "cf_workers_preview_token", &response.token
+        );
+        let websocket_url = Url::parse(&full_url)?;
+        let preview_token = response.token;
+
+        Ok(Session {
+            host,
+            websocket_url,
+            preview_token,
+        })
+    }
+}
+
+fn get_session_config(deploy_config: &DeployConfig) -> serde_json::Value {
+    match deploy_config {
+        DeployConfig::Zoned(config) => {
+            let mut routes: Vec<String> = Vec::new();
+            for route in &config.routes {
+                routes.push(route.pattern.clone());
+            }
+            json!({ "routes": routes })
+        }
+        DeployConfig::Zoneless(_) => json!({"workers_dev": true}),
+    }
+}
+
+fn get_session_address(deploy_config: &DeployConfig) -> String {
+    match deploy_config {
+        DeployConfig::Zoned(config) => format!(
+            "https://api.cloudflare.com/client/v4/zones/{}/workers/edge-preview",
+            config.zone_id
+        ),
+        // TODO: zoneless is probably wrong
+        DeployConfig::Zoneless(config) => format!(
+            "https://api.cloudflare.com/client/v4/accounts/{}/workers/subdomain/edge-preview",
+            config.account_id
+        ),
+    }
+}
+
+fn get_upload_address(target: &mut Target) -> String {
+    format!(
+        "https://api.cloudflare.com/client/v4/accounts/{}/workers/scripts/{}/edge-preview",
+        target.account_id, target.name
+    )
+}
+
+fn get_exchange_url(
+    deploy_config: &DeployConfig,
+    user: &GlobalUser,
+) -> Result<Url, failure::Error> {
+    let client = crate::http::legacy_auth_client(&user);
+    let address = get_session_address(deploy_config);
+    let url = Url::parse(&address)?;
+    let response = client.get(url).send()?.error_for_status()?;
+    let text = &response.text()?;
+    let response: SessionV4ApiResponse = serde_json::from_str(text)?;
+    let url = Url::parse(&response.result.exchange_url)?;
+    Ok(url)
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct SessionResponse {
+    pub exchange_url: String,
+    pub token: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct SessionV4ApiResponse {
+    pub result: SessionResponse,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct InspectorV4ApiResponse {
+    pub inspector_websocket: String,
+    pub token: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Preview {
+    pub preview_token: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct PreviewV4ApiResponse {
+    pub result: Preview,
+}