use std::borrow::Borrow;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::atomic::{self, AtomicU32};
use std::sync::{Arc, Mutex};

use arc_swap::ArcSwapOption;
use log::{debug, error, trace};

use tokio::net::TcpStream;
use tokio::select;
use tokio::sync::{mpsc, oneshot, watch};
use tokio_util::compat::*;

#[cfg(feature = "telemetry")]
use opentelemetry::{
	trace::{FutureExt, Span, SpanKind, TraceContextExt, Tracer},
	Context, KeyValue,
};
#[cfg(feature = "telemetry")]
use opentelemetry_contrib::trace::propagator::binary::*;

use futures::io::AsyncReadExt;

use async_trait::async_trait;

use kuska_handshake::async_std::{handshake_client, BoxStream};

use crate::endpoint::*;
use crate::error::*;
use crate::netapp::*;
use crate::proto::*;
use crate::proto2::*;
use crate::util::*;

pub(crate) struct ClientConn {
	pub(crate) remote_addr: SocketAddr,
	pub(crate) peer_id: NodeID,

	query_send: ArcSwapOption<mpsc::UnboundedSender<(RequestID, RequestPriority, Data)>>,

	next_query_number: AtomicU32,
	inflight: Mutex<HashMap<RequestID, oneshot::Sender<(Vec<u8>, AssociatedStream)>>>,
}

impl ClientConn {
	pub(crate) async fn init(
		netapp: Arc<NetApp>,
		socket: TcpStream,
		peer_id: NodeID,
	) -> Result<(), Error> {
		let remote_addr = socket.peer_addr()?;
		let mut socket = socket.compat();

		// Do handshake to authenticate and prove our identity to server
		let handshake = handshake_client(
			&mut socket,
			netapp.netid.clone(),
			netapp.id,
			netapp.privkey.clone(),
			peer_id,
		)
		.await?;

		debug!(
			"Handshake complete (client) with {}@{}",
			hex::encode(&peer_id),
			remote_addr
		);

		// Create BoxStream layer that encodes content
		let (read, write) = socket.split();
		let (mut read, write) =
			BoxStream::from_handshake(read, write, handshake, 0x8000).split_read_write();

		// Before doing anything, receive version tag and
		// check they are running the same version as us
		let mut their_version_tag = VersionTag::default();
		read.read_exact(&mut their_version_tag[..]).await?;
		if their_version_tag != netapp.version_tag {
			let msg = format!(
				"different version tags: {} (theirs) vs. {} (ours)",
				hex::encode(their_version_tag),
				hex::encode(netapp.version_tag)
			);
			error!("Cannot connect to {}: {}", hex::encode(&peer_id[..8]), msg);
			return Err(Error::VersionMismatch(msg));
		}

		// Build and launch stuff that manages sending requests client-side
		let (query_send, query_recv) = mpsc::unbounded_channel();

		let (stop_recv_loop, stop_recv_loop_recv) = watch::channel(false);

		let conn = Arc::new(ClientConn {
			remote_addr,
			peer_id,
			next_query_number: AtomicU32::from(RequestID::default()),
			query_send: ArcSwapOption::new(Some(Arc::new(query_send))),
			inflight: Mutex::new(HashMap::new()),
		});

		netapp.connected_as_client(peer_id, conn.clone());

		tokio::spawn(async move {
			let send_future = tokio::spawn(conn.clone().send_loop(query_recv, write));

			let conn2 = conn.clone();
			let recv_future = tokio::spawn(async move {
				select! {
					r = conn2.recv_loop(read) => r,
					_ = await_exit(stop_recv_loop_recv) => Ok(())
				}
			});

			send_future.await.log_err("ClientConn send_loop");

			// FIXME: should do here: wait for inflight requests to all have their response
			stop_recv_loop
				.send(true)
				.log_err("ClientConn send true to stop_recv_loop");

			recv_future.await.log_err("ClientConn recv_loop");

			// Make sure we don't wait on any more requests that won't
			// have a response
			conn.inflight.lock().unwrap().clear();

			netapp.disconnected_as_client(&peer_id, conn);
		});

		Ok(())
	}

	pub fn close(&self) {
		self.query_send.store(None);
	}

	pub(crate) async fn call<T, B>(
		self: Arc<Self>,
		rq: B,
		path: &str,
		prio: RequestPriority,
	) -> Result<<T as Message>::Response, Error>
	where
		T: Message,
		B: Borrow<T>,
	{
		let query_send = self.query_send.load_full().ok_or(Error::ConnectionClosed)?;

		// increment by 2; even are direct data; odd are associated stream
		let id = self
			.next_query_number
			.fetch_add(2, atomic::Ordering::Relaxed);
		let stream_id = id + 1;

		cfg_if::cfg_if! {
			if #[cfg(feature = "telemetry")] {
				let tracer = opentelemetry::global::tracer("netapp");
				let mut span = tracer.span_builder(format!("RPC >> {}", path))
					.with_kind(SpanKind::Client)
					.start(&tracer);
				let propagator = BinaryPropagator::new();
				let telemetry_id = Some(propagator.to_bytes(span.span_context()).to_vec());
			} else {
				let telemetry_id: Option<Vec<u8>> = None;
			}
		};

		// Encode request
		let (body, stream) = rmp_to_vec_all_named(rq.borrow())?;
		drop(rq);

		let request = QueryMessage {
			prio,
			path: path.as_bytes(),
			telemetry_id,
			body: &body[..],
		};
		let bytes = request.encode();
		drop(body);

		// Send request through
		let (resp_send, resp_recv) = oneshot::channel();
		let old = self.inflight.lock().unwrap().insert(id, resp_send);
		if let Some(old_ch) = old {
			error!(
				"Too many inflight requests! RequestID collision. Interrupting previous request."
			);
			if old_ch
				.send((vec![], Box::pin(futures::stream::empty())))
				.is_err()
			{
				debug!("Could not send empty response to collisionned request, probably because request was interrupted. Dropping response.");
			}
		}

		trace!("request: query_send {}, {} bytes", id, bytes.len());

		#[cfg(feature = "telemetry")]
		span.set_attribute(KeyValue::new("len_query", bytes.len() as i64));

		query_send.send((id, prio, Data::Full(bytes)))?;
		if let Some(stream) = stream {
			query_send.send((stream_id, prio | PRIO_SECONDARY, Data::Streaming(stream)))?;
		} else {
			query_send.send((stream_id, prio, Data::Full(Vec::new())))?;
		}

		cfg_if::cfg_if! {
			if #[cfg(feature = "telemetry")] {
				let (resp, stream) = resp_recv
					.with_context(Context::current_with_span(span))
					.await?;
			} else {
				let (resp, stream) = resp_recv.await?;
			}
		}

		if resp.is_empty() {
			return Err(Error::Message(
				"Response is 0 bytes, either a collision or a protocol error".into(),
			));
		}

		trace!("request response {}: ", id);

		let code = resp[0];
		if code == 0 {
			let ser_resp = rmp_serde::decode::from_read_ref(&resp[1..])?;
			Ok(T::Response::deserialize_msg(ser_resp, stream).await)
		} else {
			let msg = String::from_utf8(resp[1..].to_vec()).unwrap_or_default();
			Err(Error::Remote(code, msg))
		}
	}
}

impl SendLoop for ClientConn {}

#[async_trait]
impl RecvLoop for ClientConn {
	fn recv_handler(self: &Arc<Self>, id: RequestID, msg: Vec<u8>, stream: AssociatedStream) {
		trace!("ClientConn recv_handler {} ({} bytes)", id, msg.len());

		let mut inflight = self.inflight.lock().unwrap();
		if let Some(ch) = inflight.remove(&id) {
			if ch.send((msg, stream)).is_err() {
				debug!("Could not send request response, probably because request was interrupted. Dropping response.");
			}
		}
	}
}