use std::{collections::{HashMap, VecDeque}, sync::{atomic::Ordering, Arc}};

use super::{core::NodeCore, local::LocalNodeShared};
use crate::{diff, events::{Action, Event, EventEmitter, EventMap, EventMerge, EventMergeInput}, message::{Change, DataChange, Message, ObjectChange, OptionsChange, TagChange}, object::{core::{ObjectRawData, ObjectVersion}, remote::RemoteObject}, tag::remote::RemoteTag, LocalNode, ObjectDescriptor, ObjectType, TagDescriptor};

/*
This file is a bit heavy with data structures, because there is a lot to keep track of.
There is stuff that remote side has sent to us, the stuff we sent to the remote side, and there are contracts and interfaces for the local node.
On top of that there is quite a lot code at the bottom, which is entirety of the networking logic (everything above the transport layer).
*/


//TODO: document how local_node and remote_node work
//TODO: change most unwraps and panics to error returns

//TODO: consider merging this with RemoteNode
pub(super) struct State {
	options: OptionsState,
	objects: HashMap<ObjectDescriptor, ObjectState>,
	tags: HashMap<TagDescriptor, TagState>,
	object_changes: Option<EventMerge<ObjectDescriptor>>,
	tag_changes: Option<EventMerge<TagDescriptor>>,
}

#[derive(Clone)]
pub(super) struct OptionsState {
	pub(super) diff_request: Option<bool>,
}

pub(super) struct ObjectState {
	pub(super) received: RemoteObjectState,
	pub(super) sent: RemoteObjectState,
	pub(super) local: LocalObjectState,
}

pub(super) struct TagState {
	pub(super) received: RemoteTagState,
	pub(super) sent: RemoteTagState,
	pub(super) local: LocalTagState,
}

pub(super) struct RemoteObjectState {
	pub(super) observed: bool,
	pub(super) exposed: bool,
	pub(super) data: VecDeque<(/*can drop on ack*/ u64, /*version*/ ObjectVersion, /* data sent */ bool, /*data*/ ObjectRawData)>,
	pub(super) data_type: Option<ObjectType>,
	//TODO: is this redundant?
	pub(super) data_version: Option<ObjectVersion>,
	//TODO: this doesn't work
	pub(super) cost: Option<Option<u32>>,
}

pub(super) struct RemoteTagState {
	pub(super) observer: bool,
	pub(super) exposer: bool,
	pub(super) cost: Option<Option<u32>>,
}

pub(super) struct LocalObjectState {
	pub(super) object_interface: RemoteObject,
	pub(super) changes_token: Option<EventMergeInput<ObjectDescriptor>>,
}

pub(super) struct LocalTagState {
	pub(super) tag_interface: RemoteTag,
	pub(super) changes_token: Option<EventMergeInput<TagDescriptor>>,
	pub(super) objects_changes_token: Option<EventMergeInput<ObjectDescriptor>>,
}



impl State {
	pub(super) fn object_maybe_acquire(&mut self, descriptor: &ObjectDescriptor) -> Option<&mut ObjectState> {
		return if self.objects.contains_key(descriptor) { Some(self.objects.get_mut(descriptor).unwrap()) } else { None };
	}

	pub(super) fn object_acquire(&mut self, descriptor: &ObjectDescriptor, local_node: &Arc<LocalNodeShared>, node: &Arc<NodeCore>) -> &mut ObjectState {
		if !self.objects.contains_key(descriptor) {
			let mut object_state = ObjectState {
				received: RemoteObjectState { observed: false, exposed: false, data: VecDeque::new(), data_type: None, data_version: None, cost: None },
				sent: RemoteObjectState { observed: false, exposed: false, data: VecDeque::new(), data_type: None, data_version: None, cost: None },
				local: LocalObjectState { object_interface: RemoteObject::new(local_node.clone(), node.clone(), descriptor), changes_token: None },
			};
			if let Some(object_changes) = &mut self.object_changes {
				object_state.local.changes_token = Some(object_changes.include(object_state.local.object_interface.changes()));
			}
			self.objects.insert(descriptor.clone(), object_state);
		};
		self.objects.get_mut(descriptor).unwrap()
	}

	pub(super) fn object_maybe_release(&mut self, descriptor: &ObjectDescriptor) {
		let object_state = self.objects.get(descriptor).unwrap();
		if !object_state.received.observed && !object_state.received.exposed && !object_state.sent.observed && !object_state.sent.exposed {
			let mut object_state = self.objects.remove(descriptor).unwrap();
			drop(object_state.local.changes_token.take());
		}
	}

	pub(super) fn tag_maybe_acquire(&mut self, descriptor: &TagDescriptor) -> Option<&mut TagState> {
		return if self.tags.contains_key(descriptor) { Some(self.tags.get_mut(descriptor).unwrap()) } else { None };
	}

	pub(super) fn tag_acquire(&mut self, descriptor: &TagDescriptor, local_node: &Arc<LocalNodeShared>, node: &Arc<NodeCore>) -> &mut TagState {
		if !self.tags.contains_key(descriptor) {
			let mut tag_state = TagState {
				received: RemoteTagState { observer: false, exposer: false, cost: None },
				sent: RemoteTagState { observer: false, exposer: false, cost: None },
				local: LocalTagState {
					tag_interface: RemoteTag::new(local_node.clone(), node.clone(), descriptor),
					changes_token: None,
					objects_changes_token: None,
				},
			};
			if let Some(tag_changes) = &mut self.tag_changes {
				tag_state.local.changes_token = Some(tag_changes.include(tag_state.local.tag_interface.changes()));
			}
			if let Some(object_changes) = &mut self.object_changes {
				tag_state.local.objects_changes_token = Some(object_changes.include(tag_state.local.tag_interface.objects_changes()));
			}
			self.tags.insert(descriptor.clone(), tag_state);
		};
		self.tags.get_mut(descriptor).unwrap()
	}

	pub(super) fn tag_maybe_release(&mut self, descriptor: &TagDescriptor) {
		let tag_state = self.tags.get(descriptor).unwrap();
		if !tag_state.received.observer && !tag_state.received.exposer && !tag_state.sent.observer && !tag_state.sent.exposer {
			let mut tag_state = self.tags.remove(descriptor).unwrap();
			drop(tag_state.local.changes_token.take());
			drop(tag_state.local.objects_changes_token.take());
		}
	}
}

impl RemoteObjectState {
	pub(super) fn apply_change(&mut self, change: &ObjectChange) {
		self.observed = change.observer.unwrap_or(self.observed);
		self.exposed = change.exposer.unwrap_or(self.exposed);
		self.data_version = change.data_version.clone().or_else(|| self.data_version.take()); //because I hate "if"
		self.data_type = change.data_type.clone().or_else(|| self.data_type.take());
		self.cost = change.cost.or(self.cost);
	}
}

impl RemoteTagState {
	pub(super) fn apply_change(&mut self, change: &TagChange) {
		self.observer = change.observer.unwrap_or(self.observer);
		self.exposer = change.exposer.unwrap_or(self.exposer);
		//TODO: cost? self.cost = change.cost.or(self.cost);
	}
}

#[derive(Clone, Eq, PartialEq, Hash, Debug)]
pub enum Descriptor {
	Object(ObjectDescriptor),
	Tag(TagDescriptor),
}

pub struct RemoteNode {
	local_node: Arc<LocalNodeShared>,
	//TODO: de-pub, and implement accessor
	pub local_node_events: EventMerge<Descriptor>,
	//TODO: remote should die when nodecore dies
	node_core: Arc<NodeCore>,
	state: State,
	// last sent message number
	message_sequence_number: u64,
	// last received message number
	message_acknowledgment_number: u64,
	diff_produce: Option<bool>,
	diff_request: Option<bool>,
}


impl Drop for RemoteNode {
	fn drop(&mut self) {
		//FORMER-BUG, objects and tags clear was missing, leading to wrong ref counts
		//REGRESSION-TEST-NEEDED
		self.state.objects.clear();
		self.state.tags.clear();
		assert_eq!(Arc::strong_count(&self.node_core), 1);
	}
}



impl RemoteNode {
	pub fn new(local_node: &LocalNode, upstream: bool, diff_produce: Option<bool>, diff_request: Option<bool>) -> RemoteNode {
		let local_node_events = EventMerge::new();
		let object_events_map = EventMap::new(EventMerge::new(), Descriptor::Object);
		let tag_events_map = EventMap::new(EventMerge::new(), Descriptor::Tag);
		let mut object_changes = None;
		let mut tag_changes = None;

		if upstream {
			object_events_map.input().include(local_node.shared.objects_changes()).forever();
			tag_events_map.input().include(local_node.shared.tags_changes()).forever();
		} else {
			object_changes = Some(object_events_map.input().clone());
			tag_changes = Some(tag_events_map.input().clone());
		};
		local_node_events.include(object_events_map).forever();
		local_node_events.include(tag_events_map).forever();

		let state = State { objects: HashMap::new(), tags: HashMap::new(), object_changes, tag_changes, options: OptionsState { diff_request: None } };
		RemoteNode {
			node_core: NodeCore::new(upstream, Some(0), Some(0), None),
			local_node: local_node.shared.clone(),
			local_node_events,
			state,
			message_sequence_number: 0,
			message_acknowledgment_number: 0,
			diff_produce,
			diff_request,
		}
	}

	//TODO: some kind of id() maybe too
	pub fn name(&self) -> String {
		self.node_core.name.read().unwrap().clone().unwrap_or_else(|| "-".to_string())
	}

	pub fn connected(&mut self) -> Result<Option<Message>, String> {
		let mut changes_to_send = vec![Change::Options(OptionsChange {
			name: self.local_node.node_core.as_ref().unwrap().name.read().unwrap().clone(),
			custom: None,
			expose_capacity: Some(self.local_node.node_core.as_ref().unwrap().capacity.load(Ordering::Relaxed)),
			expose_load_limit: Some(self.local_node.node_core.as_ref().unwrap().limit.load(Ordering::Relaxed)),
			diff_request: Some(self.diff_request),
		})];
		for key in self.local_node_events.keys() {
			changes_to_send.extend(match key {
				Descriptor::Object(descriptor) => self.process_object_event(descriptor, Action::Change)?,
				Descriptor::Tag(descriptor) => self.process_tag_event(descriptor, Action::Change)?,
			});
		}
		Ok(self.build_message(changes_to_send))
	}

	pub fn disconnected(&mut self) {
		self.state.objects.clear();
		self.state.tags.clear();
		self.state.options.diff_request = None;
		self.message_sequence_number = 0;
		self.message_acknowledgment_number = 0;
	}

	//TODO replace String error with proper error enum
	pub fn received_message(&mut self, message: Message) -> Result<Option<Message>, String> {
		let changes_to_send = vec![];
		self.message_acknowledgment_number = message.sequence_number;

		for change in message.changes {
			match change {
				Change::Options(options_change) => {
					if let Some(diff_request) = options_change.diff_request {
						self.state.options.diff_request = diff_request;
					};
					if let Some(name) = options_change.name {
						self.node_core.name_set(&name);
					};
					if let Some(capacity) = options_change.expose_capacity {
						self.node_core.capacity.store(capacity, Ordering::SeqCst);
						//TODO: rebalance
					}
					if let Some(limit) = options_change.expose_load_limit {
						self.node_core.limit.store(limit, Ordering::SeqCst);
						//TODO: rebalance
					}
					if let Some(diff_request) = options_change.diff_request {
						self.state.options.diff_request = diff_request;
					}
					//TODO: custom data
				}

				Change::Object(object_change) => {
					let object_state = self.state.object_acquire(&object_change.descriptor, &self.local_node, &self.node_core);
					if let Some(observer) = object_change.observer {
						if observer == object_state.received.observed {
							return Err("Protocol error. Unneeded data received (object observer field).".to_string());
						};
						object_state.local.object_interface.observe(observer);
						object_state.received.observed = observer;
					}
					if let Some(exposer) = object_change.exposer {
						if exposer == object_state.received.exposed {
							return Err("Protocol error. Unneeded data received (object exposer field).".to_string());
						};
						let cost = if exposer { object_change.cost.flatten().or_else(|| object_state.received.cost.flatten()).expect("No cost?") } else { 0 };
						object_state.local.object_interface.expose(exposer, cost);
						object_state.received.exposed = exposer;
					}
					if let Some(data_version) = &object_change.data_version {
						if let Some(data) = object_change.data {
							match data {
								DataChange::Data { data } => {
									object_state.received.data.clear();
									object_state.received.data.push_back((message.sequence_number, data_version.clone(), true, data));
								}
								DataChange::Diff { diff, old_data_version } => {
									let empty = Arc::new(vec![]);
									let old_data = if let Some(old_data_version) = old_data_version {
										if let Some((_sequence, _version, _data_present, data)) =
											object_state.sent.data.iter().find(|(_sequence, version, _data_present, _data)| *version == old_data_version)
										{
											data.clone()
										} else if let Some((_sequence, version, _data_present, data)) = object_state.received.data.pop_back() {
											if *version == old_data_version {
												data
											} else {
												return Err("CAKE not detected in receptacle(1)".to_string());
											}
										} else {
											//dbg!(old_data_version);
											return Err("CAKE not detected in receptacle(2)".to_string());
										}
									} else {
										empty
									};
									object_state.received.data.clear();
									object_state.received.data.push_back((
										message.sequence_number,
										data_version.clone(),
										true,
										Arc::new(diff::patch(&old_data, &diff)?),
									));
								}
							}
						};
						if let Some(data_type) = object_change.data_type {
							object_state.received.data_type = Some(data_type);
						};
						if !object_state.received.data.is_empty() && object_state.received.data_type.is_some() {
							if object_state.local.object_interface.is_exposed() {
								//TODO: or should we check if node is active exposer?
								object_state.local.object_interface.state_set(
									data_version,
									object_state.received.data.back().as_ref().unwrap().3.clone(),
									object_state.received.data_type.as_ref().unwrap().clone(),
								);
							} else {
								return Err("Remote state error. Data submitted for object remote is not a exposer of.".to_string());
							}
						} else {
							//TODO: if there is no data sent, should that only be permitted for initial message? only for observers?
							object_state.local.object_interface.maybe_synchronize(data_version);
						}
						object_state.received.data_version = object_change.data_version;
					} else if object_change.data.is_some() || object_change.data_type.is_some() {
						return Err("Remote state error. Data or data type submitted without data version.".to_string());
					}
					self.state.object_maybe_release(&object_change.descriptor);
				}

				Change::Tag(tag_change) => {
					let tag_state = self.state.tag_acquire(&tag_change.descriptor, &self.local_node, &self.node_core);
					if let Some(observer) = tag_change.observer {
						if observer == tag_state.received.observer {
							return Err("Protocol error. Unneeded data received (tag observer field).".to_string());
						};
						tag_state.local.tag_interface.observe(observer);
						tag_state.received.observer = observer;
					};
					if let Some(exposer) = tag_change.exposer {
						if exposer == tag_state.received.exposer {
							return Err("Protocol error. Unneeded data received (tag exposer field).".to_string());
						};
						let cost = if exposer { tag_change.cost.flatten().or_else(|| tag_state.received.cost.flatten()).expect("No cost?") } else { 0 };
						tag_state.local.tag_interface.expose(exposer, cost);
						tag_state.received.exposer = exposer;
					};
					self.state.tag_maybe_release(&tag_change.descriptor);
				}
			}
		}

		for object_state in self.state.objects.values_mut() {
			while object_state.sent.data.len() > 1 && object_state.sent.data.front().unwrap().0 <= message.acknowledgment_number {
				object_state.sent.data.pop_front();
			}
		}

		Ok(self.build_message(changes_to_send))
	}

	pub fn received_local_node_event(&mut self, event: Event<Descriptor>) -> Result<Option<Message>, String> {
		let changes = match event.key {
			Descriptor::Object(descriptor) => self.process_object_event(descriptor, event.action)?,
			Descriptor::Tag(descriptor) => self.process_tag_event(descriptor, event.action)?,
		};
		Ok(self.build_message(changes))
	}

	fn build_message(&mut self, changes: Vec<Change>) -> Option<Message> {
		if changes.is_empty() {
			None
		} else {
			self.message_sequence_number += 1;
			Some(Message { sequence_number: self.message_sequence_number, acknowledgment_number: self.message_acknowledgment_number, changes })
		}
	}

	fn process_object_event(&mut self, descriptor: ObjectDescriptor, action: Action) -> Result<Vec<Change>, String> {
		match action {
			Action::Insert | Action::Change => {
				let options = self.state.options.clone();
				let object_state = self.state.object_acquire(&descriptor, &self.local_node, &self.node_core);

				let last_received_version = object_state.received.data_version.clone();
				let last_sent_version = object_state.sent.data_version.clone();
				let highest_version_known_by_remote = vec![last_received_version.as_ref(), last_sent_version.as_ref()].into_iter().flatten().max();

				let (_synchronized, data_version, data_type, data) = object_state.local.object_interface.state();
				let local_node_knows_newer_version_than_remote = data_version
					.as_ref()
					.and_then(|newest_version| highest_version_known_by_remote.as_ref().map(|known_version| newest_version > known_version).or(Some(true)))
					.unwrap_or(false);

				let (declare_local_as_observer, declare_local_as_exposer) = if self.node_core.upstream {
					(object_state.local.object_interface.has_other_observer(false), object_state.local.object_interface.has_other_exposer(false))
				} else {
					(
						object_state.local.object_interface.is_assigned() && object_state.local.object_interface.has_other_observer(true),
						!object_state.local.object_interface.is_assigned()
							&& object_state.local.object_interface.has_other_exposer(true)
							&& object_state.local.object_interface.is_observed(),
					)
				};

				let declare_cost = if declare_local_as_exposer { object_state.local.object_interface.minimum_cost_in_other_nodes() } else { None };

				let declare_data =
					//TODO: rethink this condition
					if object_state.local.object_interface.has_other_exposer(true) && object_state.local.object_interface.is_observed() && local_node_knows_newer_version_than_remote {
						if self.diff_produce.unwrap_or_else(|| options.diff_request.unwrap_or(true)) {
							//TODO: this is messy af
							let (newest_shared_version, newest_shared_data) = match (last_received_version, last_sent_version) {
								(None, None) => (None, None),
								(Some(received_version), None) =>
									if object_state.received.data.back().is_some() {
										(Some(received_version), object_state.received.data.back())
									} else {
										(None, None)
									},
								(None, Some(sent_version)) =>
									if let Some(sent) = object_state.sent.data.back() {
										if sent.2 {
											(Some(sent_version), Some(sent))
										} else {
											(None, None)
										}
									} else {
										(None, None)
									},
								(Some(received_version), Some(sent_version)) =>
									if received_version > sent_version && object_state.received.data.back().is_some() {
										(Some(received_version), object_state.received.data.back())
									} else if let Some(sent) = object_state.sent.data.back() {
										if sent.2 {
											(Some(sent_version), Some(sent))
										} else {
											(None, None)
										}
									} else {
										(None, None)
									},
							};
							let diff = object_state.local.object_interface.get_diff(
								&newest_shared_version,
								newest_shared_data.map(|data| &data.3),
								data_version.as_ref().unwrap(),
								data.as_ref().unwrap(),
							);
							if let Some(diff) = diff {
								Some(DataChange::Diff { diff, old_data_version: newest_shared_version })
							} else {
								Some(DataChange::Data { data: data.clone().unwrap() })
							}
						} else {
							Some(DataChange::Data { data: data.clone().unwrap() })
						}
					} else {
						None
					};

				let change = ObjectChange {
					descriptor: descriptor.clone(),
					observer: if declare_local_as_observer != object_state.sent.observed { Some(declare_local_as_observer) } else { None },
					exposer: if declare_local_as_exposer != object_state.sent.exposed { Some(declare_local_as_exposer) } else { None },
					data_type: if declare_data.is_some() && data_type != object_state.sent.data_type { Some(data_type.unwrap()) } else { None },
					data_version: if declare_data.is_some()
						|| ((declare_local_as_observer && !object_state.sent.observed) || (declare_local_as_exposer && !object_state.sent.exposed))
					{
						data_version.clone()
					} else {
						None
					},
					data: declare_data,
					cost: if Some(declare_cost) != object_state.sent.cost { Some(declare_cost) } else { None },
				};
				object_state.sent.apply_change(&change);

				if change.is_empty() {
					Ok(vec![])
				} else {
					if change.data_version.is_some() {
						object_state.sent.data.push_back((self.message_sequence_number, data_version.unwrap(), change.data.is_some(), data.unwrap()));
					}
					Ok(vec![Change::Object(change)])
				}
			}
			Action::Remove => {
				if let Some(object_state) = self.state.object_maybe_acquire(&descriptor) {
					let change = ObjectChange {
						descriptor: descriptor.clone(),
						observer: if object_state.sent.observed { Some(false) } else { None },
						exposer: if object_state.sent.exposed { Some(false) } else { None },
						//TODO: should we clear all following fields?
						data: None,
						data_type: None,
						data_version: None,
						cost: None,
					};

					object_state.sent.apply_change(&change);
					self.state.object_maybe_release(&descriptor);

					if !change.is_empty() {
						Ok(vec![Change::Object(change)])
					} else {
						Ok(vec![])
					}
				} else {
					Ok(vec![])
				}
			}
		}
	}

	fn process_tag_event(&mut self, descriptor: TagDescriptor, action: Action) -> Result<Vec<Change>, String> {
		if self.node_core.upstream {
			match action {
				Action::Insert | Action::Change => {
					let tag_state = self.state.tag_acquire(&descriptor, &self.local_node, &self.node_core);

					let declare_local_as_observer = tag_state.local.tag_interface.has_other_observer();
					let declare_local_as_exposer = tag_state.local.tag_interface.has_other_exposer();
					let declare_cost = if declare_local_as_exposer { tag_state.local.tag_interface.minimum_cost_in_other_nodes() } else { None };

					let change = TagChange {
						descriptor: descriptor.clone(),
						observer: if declare_local_as_observer != tag_state.sent.observer { Some(declare_local_as_observer) } else { None },
						exposer: if declare_local_as_exposer != tag_state.sent.exposer { Some(declare_local_as_exposer) } else { None },
						cost: if Some(declare_cost) != tag_state.sent.cost { Some(declare_cost) } else { None },
					};

					tag_state.sent.apply_change(&change);
					self.state.tag_maybe_release(&descriptor);

					Ok(if !change.is_empty() { vec![Change::Tag(change)] } else { vec![] })
				}
				Action::Remove =>
					if let Some(tag_state) = self.state.tag_maybe_acquire(&descriptor) {
						let change = TagChange {
							descriptor: descriptor.clone(),
							observer: if tag_state.sent.observer { Some(false) } else { None },
							exposer: if tag_state.sent.exposer { Some(false) } else { None },
							//TODO: should we clear this explicitly?
							cost: None,
						};
						self.state.tag_maybe_release(&descriptor);
						Ok(if !change.is_empty() { vec![Change::Tag(change)] } else { vec![] })
					} else {
						Ok(vec![])
					},
			}
		} else {
			Ok(vec![])
		}
	}
}