sark 0.9.0

use std::io;
use std::net::SocketAddr;

use cartel_redis::{Connect, DEFAULT_BACKOFF, Ops, Redis};
use dope::manifold::connector::source::Static;
use dope::manifold::env::Bundle;
use dope_net::tcp::Tcp;
use dope_net::wire::identity::Identity;
use http::StatusCode;
use sark::{HttpServer, Throughput, app, driver, listener, tcp};

type Env = Bundle<Tcp, Identity, Throughput>;

mod support;

const MAX_CONNECTIONS: usize = 1024;
const HTTP_LISTENER_ID: u8 = 0;
const DATE_UPDATER_ID: u8 = 1;
const REDIS_CONNECTOR_ID: u8 = 2;
const VISIT_KEYS: [&[u8]; 1] = [b"sark:visits"];
#[derive(Clone, Copy)]
struct AppState<'d> {
    redis: Redis<'d>,
}

#[sark_gen::json(encode)]
struct VisitBody {
    ok: bool,
    visits: i64,
    error: String,
}

#[sark_gen::response(json)]
#[header("content-type", "application/json")]
struct VisitResponse {
    status: StatusCode,
    body: VisitBody,
}

fn response(status: StatusCode, body: VisitBody) -> VisitResponse {
    VisitResponse { status, body }
}

#[sark_gen::handler]
async fn visit(state: &AppState<'_>) -> VisitResponse {
    match state.redis.incr(b"sark:visits").await {
        Ok(visits) => response(
            StatusCode::OK,
            VisitBody {
                ok: true,
                visits,
                error: String::new(),
            },
        ),
        Err(error) => response(
            StatusCode::BAD_GATEWAY,
            VisitBody {
                ok: false,
                visits: 0,
                error: error.to_string(),
            },
        ),
    }
}

#[sark_gen::handler]
async fn reset(state: &AppState<'_>) -> VisitResponse {
    match state.redis.del(&VISIT_KEYS).await {
        Ok(_) => response(
            StatusCode::OK,
            VisitBody {
                ok: true,
                visits: 0,
                error: String::new(),
            },
        ),
        Err(error) => response(
            StatusCode::BAD_GATEWAY,
            VisitBody {
                ok: false,
                visits: 0,
                error: error.to_string(),
            },
        ),
    }
}

sark_gen::define_route! {
    HttpWithRedisApp: AppState<'_> => {
        GET "/" => async(capacity = MAX_CONNECTIONS) visit,
        POST "/reset" => async(capacity = MAX_CONNECTIONS) reset,
    }
}

fn main() -> io::Result<()> {
    let bind: SocketAddr = std::env::var("BIND")
        .unwrap_or_else(|_| "127.0.0.1:8080".to_string())
        .parse()
        .expect("invalid BIND");
    let redis_addr: SocketAddr = std::env::var("REDIS_ADDR")
        .unwrap_or_else(|_| "127.0.0.1:6379".to_string())
        .parse()
        .expect("invalid REDIS_ADDR");
    let server = HttpServer::<HTTP_LISTENER_ID, DATE_UPDATER_ID, Throughput>::new(
        listener::Config::<Tcp> {
            bind,
            max_connections: MAX_CONNECTIONS,
            backlog: 1024,
            stream: tcp::stream::Config {
                no_delay: Some(true),
                ..Default::default()
            },
            transport: tcp::listener::Config {
                reuse_port: true,
                ..Default::default()
            },
            egress: Default::default(),
        },
        std::time::Duration::from_secs(10),
    );

    eprintln!("sark redis: listening on http://{bind}, upstream redis {redis_addr}");
    let redis_factory = support::redis_config().map_err(io::Error::other)?.factory();

    server.run_with_storage(
        vec![0u16],
        |_| driver::Config::for_tcp_profile::<Throughput>(MAX_CONNECTIONS),
        move |_, _| redis_factory,
        move |server, session| {
            let backoff = session
                .seed()
                .derive(dope::hash::domain::BACKOFF ^ REDIS_CONNECTOR_ID as u64)
                .state();
            let store = session.storage() as *const cartel_redis::Store<'_>;
            // The store remains pinned in executor storage until after the
            // connector resource and HTTP application are dropped.
            let redis = unsafe { (&*store).redis() };
            let connector = {
                let mut driver = session.driver_access();
                redis.connect::<REDIS_CONNECTOR_ID, _, Env>(
                    Connect {
                        topology: Static::<Tcp>::new(vec![redis_addr], DEFAULT_BACKOFF, backoff),
                    },
                    &mut driver,
                )?
            };
            let state = AppState { redis };
            let app = HttpWithRedisApp::new(
                state,
                app::Config {
                    timer_capacity: MAX_CONNECTIONS.saturating_mul(2),
                    task_capacity: MAX_CONNECTIONS,
                },
            );
            server.serve_with_resource(session, app, connector, None)
        },
    )
}