use fastly::{Error, Request};
use lazy_static::lazy_static;
use regex::Regex;
use serde_json::Value;
use std::io::Write;

lazy_static! {
    // Regex to capture key value pair of format "key: value"
    static ref DATA_REGEX: Regex = Regex::new(r"^([^:]+?)\s*:\s*(.*?)\s*$").unwrap();
}

// Allow list of cities whoes information will be emitted to the client
const ALLOW_CITY_LIST: [&str; 10] = [
    "Atlanta",
    "Berlin",
    "Dublin",
    "Boston",
    "Denver",
    "Tokyo",
    "Singapore",
    "Los Angeles",
    "Dubai",
    "San Francisco",
];

// The delimiter that seperate events
const DELIMITER: &str = "\n\n";

// Chunk size of each origin response read
const CHUNK_SIZE: usize = 1024;

fn main() -> Result<(), Error> {
    let req = Request::from_client();
    let mut backend_resp = req.send("origin_0")?;

    // Take the body so we can iterate through its lines later
    let mut backend_resp_body = backend_resp.take_body();

    // Start sending the backend response to the client with a now-empty body
    let mut client_body = backend_resp.stream_to_client();

    let mut buffer = String::new();
    for chunk in backend_resp_body.read_chunks(CHUNK_SIZE) {
        let Ok(chunk) = chunk else {
            // Abort operation if encountered any read error
            break;
        };

        let mut events_stream = String::from_utf8_lossy(&chunk);
        if !buffer.is_empty() {
            events_stream = format!("{}{}", &buffer, &events_stream).into();
        }

        // Break event stream into events
        let mut events: Vec<&str> = events_stream.split(DELIMITER).collect();

        // Save the last event in a buffer for next round of parsing
        buffer = events.pop().unwrap_or_default().to_string();

        // Filter events and emit
        for event in events {
            if let Some(city) = get_event_city(event) {
                let allow = ALLOW_CITY_LIST.contains(&city.as_str());
                println!("City: {city}, include: {allow}");
                if !allow {
                    // Skip the city that is not in the allow list
                    continue;
                }
            }

            // Emit the event to client
            client_body.write_str(event);
            client_body.write_str(DELIMITER);

            // Flush the stream so that client can receive the event faster
            client_body.flush().ok();
        }
    }

    // Finish the streaming body to close the client connection
    client_body.finish().ok();

    Ok(())
}

// Get destination city name of the event
fn get_event_city(event: &str) -> Option<String> {
    event.split('\n').find_map(|line| {
        let cap = DATA_REGEX.captures(line)?;
        let key = cap.get(1)?.as_str();

        // We only care about key "data", which is followed by a value of JSON object
        if key != "data" {
            return None;
        }

        let value = cap.get(2)?.as_str();

        // Parse JSON value of key "data"
        let json_v = serde_json::from_str::<Value>(value).ok()?;

        // Get destination city
        Some(json_v["destination"].as_str()?.to_string())
    })
}