use std::fmt::Write as _;
use std::io::Write as _;

fn env(name: &str) -> Option<String> {
	match std::env::var(name) {
		Ok(v) => Some(v),
		Err(std::env::VarError::NotPresent) => None,
		Err(err) => panic!("Could not read {}: {}", name, err),
	}
}

struct Config {
	batch: u32,
	batch_size: usize,
	cursor_path: std::path::PathBuf,
	debug: bool,
	excludes: Vec<(String, Vec<u8>)>,
	method: reqwest::Method,
	url: url::Url,
}

impl Config {
	fn excludes_from_env() -> Vec<(String, Vec<u8>)> {
		let mut excludes = Vec::new();
		let mut key = "JF_EXCLUDE_999".to_string();
		let key_base = key.len() - 3;
		for i in 0.. {
			key.truncate(key_base);
			write!(&mut key, "{}", i).unwrap();

			let Some(filter) = std::env::var_os(&key)
				else { break };

			let equals = filter.as_encoded_bytes().iter()
				.take_while(|&&c| c != b'=')
				.count();

			if equals == 0 {
				panic!(
					r#"{:?} must be formatted as KEY=VALUE, no "=" in {:?}"#,
					key,
					filter.to_string_lossy());
			}

			let key = &filter.as_encoded_bytes()[..equals];
			let key = std::str::from_utf8(key).expect("JF_EXCLUDE key is not valid UTF-8");

			let value = &filter.as_encoded_bytes()[(equals+1)..];

			excludes.push((
				key.into(),
				value.into(),
			));
		}
		excludes
	}

	fn from_env() -> Self {
		if env("JF_FILTER").is_some() {
			panic!("JF_FILTER is not supported in the Rust version.")
		}
		Config {
			debug: env("JF_DEBUG")
				.map(|v| !v.is_empty())
				.unwrap_or(false),
			batch: env("JF_BATCH")
				.map(|n| n.parse().expect("JF_BATCH is not an integer"))
				.unwrap_or(100),
			batch_size: env("JF_BATCH_SIZE")
				.map(|n| n.parse().expect("JF_BATCH_SIZE is not an integer"))
				.unwrap_or(10 * 1024 * 1024),
			cursor_path: env("JF_CURSOR")
				.unwrap_or_else(|| "journal-forwarder.cursor".into())
				.into(),
			excludes: Self::excludes_from_env(),
			method: env("JF_METHOD")
				.map(|m| m.parse().expect("JF_METHOD is invalid"))
				.unwrap_or(reqwest::Method::POST),
			url: env("JF_URL")
				.or_else(|| env("JF_URL_SRC")
					.map(|path| std::fs::read_to_string(path).expect("Could not read JF_URL_SRC")))
				.expect("JF_URL or JF_URL_SRC env variables are required.")
				.parse().expect("JF_URL is not a valid URL"),
		}
	}
}

fn log(
	message_id: impl Into<String>,
	message: impl ToString,
	priority: u8,
	fields: impl IntoIterator<Item=(String, String)>,
) {
	let mut e = journald::JournalEntry::new();
	e.fields.insert("MESSAGE_ID".into(), message_id.into());
	e.fields.insert("MESSAGE".into(), message.to_string());
	e.fields.insert("PRIORITY".into(), priority.to_string());
	e.fields.insert("SYSLOG_IDENTIFIER".into(), "journal-forwarder".into());
	e.fields.extend(fields);
	if let Err(err) = journald::writer::submit(&e) {
		eprintln!("{:?}", e);
		eprintln!("Failed to write to the journal: {}", err);
	}
}

fn main() {
	let config = Config::from_env();

	let mut journal = journald::reader::JournalReader::open(&Default::default()).unwrap();
	let mut cursor_read = String::new();

	match std::fs::read_to_string(&config.cursor_path) {
		Ok(c) => {
			cursor_read.clone_from(&c);
			if let Err(err) = journal.seek(journald::reader::JournalSeek::Cursor(c)) {
				log(
					"75a7247ca3324431b039a3d66ca39543",
					"Invalid cursor. Falling back to current time, messages may be missed.",
					3,
					[
						("INVALID_CURSOR".to_string(), cursor_read.clone()),
						("JOURNALD_RETURNED".into(), err.to_string()),
					]);
				journal.seek(journald::reader::JournalSeek::Tail).unwrap();
			}
		},
		Err(err) => match err.kind() {
			std::io::ErrorKind::NotFound => {
				log(
					"f2e153ec33cc4037a9ca1a4180a598de",
					"This appears to be the first run, running from start.",
					5,
					[]);
				journal.seek(journald::reader::JournalSeek::Head).unwrap();
			}
			err => panic!("Error reading saved cursor: {}", err)
		}
	}

	let client = reqwest::blocking::Client::builder()
		.user_agent("journal-forwarder.kevincox.ca/1")
		.connect_timeout(std::time::Duration::from_secs(60))
		.timeout(std::time::Duration::from_secs(10 * 60))
		.build().unwrap();

	log(
		"4f56c3e133bc411383d7200165e2e866",
		format!("Starting with cursor {:?}", cursor_read),
		6,
		[("START_CURSOR".to_string(), cursor_read.clone())]);

	let mut batch = Vec::new();
	let mut cursor_write = String::new();

	loop {
		let mut batch_end = None;

		'entry: for _ in 0..config.batch {
			let entry = journal.next_entry()
				.expect("Could not read entry from journal.");
			let entry = match entry {
				Some(e) => e,
				None => break,
			};

			for (key, value) in &config.excludes {
				let Some(entry_value) = entry.fields.get(key)
					else { continue };
				if entry_value.as_bytes() == value {
					continue 'entry
				}
			}

			let prev_size = batch.len();

			serde_json::to_writer(&mut batch, &entry.fields).unwrap();
			batch.push(b'\n');

			if batch.len() > config.batch_size && !batch.is_empty() {
				cursor_write.clone_from(&cursor_read);
				batch_end = Some(prev_size);
			}

			cursor_read.clear();
			cursor_read.push_str(entry.get_field("__CURSOR").unwrap());

			if batch_end.is_some() { break }
			if batch.len() == config.batch_size { break }
		}

		if batch.is_empty() {
			journal.wait().expect("Failed to wait for new entries.");
			continue
		}

		let batch_cursor = match batch_end {
			Some(_) => &cursor_write,
			None => &cursor_read,
		};
		let batch_range = ..(batch_end.unwrap_or(batch.len()));
		let chunk = &batch[batch_range];

		if config.debug {
			println!(" --- BATCH BOUNDARY ---");
			std::io::stdout().write_all(&chunk).unwrap();
		} else {
			client.request(config.method.clone(), config.url.clone())
				.header("content-type", "application/json")
				.body(chunk.to_vec())
				.send()
				.expect("Failed to send logs.");
		}

		write_atomic::write_file(&config.cursor_path, batch_cursor.as_bytes())
			.expect("failed to checkpoint cursor");

		batch.drain(batch_range);
	}
}
