use std::ops::ControlFlow;

use bestool_postgres::error::{format_db_error, format_mobc_error};
use comfy_table::Table;

use crate::repl::state::ReplContext;

use super::output::OutputWriter;

pub(super) async fn handle_describe_table(
	ctx: &mut ReplContext<'_>,
	schema: &str,
	table_name: &str,
	detail: bool,
	sameconn: bool,
	writer: &OutputWriter,
) -> ControlFlow<()> {
	// Detect PostgreSQL version to handle version-specific columns
	let version_query = "SHOW server_version_num";
	let version_result = if sameconn {
		ctx.client.query(version_query, &[]).await
	} else {
		match ctx.pool.get().await {
			Ok(client) => client.query(version_query, &[]).await,
			Err(e) => {
				eprintln!(
					"Error getting connection from pool: {}",
					format_mobc_error(&e, None)
				);
				return ControlFlow::Continue(());
			}
		}
	};

	let server_version: i32 = match version_result {
		Ok(rows) => {
			if let Some(row) = rows.first() {
				let version_str: String = row.get(0);
				version_str.parse().unwrap_or(0)
			} else {
				0
			}
		}
		Err(_) => 0,
	};

	// PostgreSQL 14 introduced the attcompression column
	let compression_column = if server_version >= 140000 {
		"COALESCE(a.attcompression::text, '')"
	} else {
		"''"
	};

	let columns_query = format!(
		r#"
		SELECT
			a.attname AS column_name,
			pg_catalog.format_type(a.atttypid, a.atttypmod) AS data_type,
			CASE WHEN a.attnotnull THEN 'not null' ELSE '' END AS nullable,
			CASE
				WHEN pg_catalog.pg_get_expr(d.adbin, d.adrelid, true) IS NOT NULL
					THEN 'default ' || pg_catalog.pg_get_expr(d.adbin, d.adrelid, true)
				ELSE ''
			END AS default_value,
			CASE
				WHEN a.attidentity = 'a' THEN 'generated always as identity'
				WHEN a.attidentity = 'd' THEN 'generated by default as identity'
				ELSE ''
			END AS identity,
			CASE WHEN a.attgenerated = 's' THEN 'generated' ELSE '' END AS generated,
			pg_catalog.col_description(c.oid, a.attnum) AS description,
			CASE
				WHEN co.collname IS NOT NULL THEN co.collname
				ELSE ''
			END AS collation,
			{} AS compression,
			CASE
				WHEN a.attstorage = 'p' THEN 'plain'
				WHEN a.attstorage = 'e' THEN 'external'
				WHEN a.attstorage = 'm' THEN 'main'
				WHEN a.attstorage = 'x' THEN 'extended'
				ELSE ''
			END AS storage
		FROM pg_catalog.pg_class c
		LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
		LEFT JOIN pg_catalog.pg_attribute a ON a.attrelid = c.oid
		LEFT JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
		LEFT JOIN pg_catalog.pg_attrdef d ON d.adrelid = c.oid AND d.adnum = a.attnum
		LEFT JOIN pg_catalog.pg_collation co ON co.oid = a.attcollation AND a.attcollation <> t.typcollation
		WHERE n.nspname = $1
			AND c.relname = $2
			AND a.attnum > 0
			AND NOT a.attisdropped
		ORDER BY a.attnum
	"#,
		compression_column
	);

	let indexes_query = r#"
		SELECT
			i.relname AS index_name,
			pg_catalog.pg_get_indexdef(i.oid) AS index_def,
			CASE WHEN ix.indisprimary THEN 'PRIMARY KEY'
				WHEN ix.indisunique THEN 'UNIQUE'
				ELSE ''
			END AS constraint_type,
			am.amname AS index_type
		FROM pg_catalog.pg_index ix
		LEFT JOIN pg_catalog.pg_class i ON i.oid = ix.indexrelid
		LEFT JOIN pg_catalog.pg_class c ON c.oid = ix.indrelid
		LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
		LEFT JOIN pg_catalog.pg_am am ON i.relam = am.oid
		WHERE n.nspname = $1
			AND c.relname = $2
		ORDER BY ix.indisprimary DESC, ix.indisunique DESC, i.relname
	"#;

	let foreign_keys_query = r#"
		SELECT
			conname AS constraint_name,
			pg_catalog.pg_get_constraintdef(oid, true) AS constraint_def
		FROM pg_catalog.pg_constraint
		WHERE conrelid = (
			SELECT c.oid
			FROM pg_catalog.pg_class c
			LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
			WHERE n.nspname = $1 AND c.relname = $2
		)
		AND contype = 'f'
		ORDER BY conname
	"#;

	let check_constraints_query = r#"
		SELECT
			conname AS constraint_name,
			pg_catalog.pg_get_constraintdef(oid, true) AS constraint_def
		FROM pg_catalog.pg_constraint
		WHERE conrelid = (
			SELECT c.oid
			FROM pg_catalog.pg_class c
			LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
			WHERE n.nspname = $1 AND c.relname = $2
		)
		AND contype = 'c'
		ORDER BY conname
	"#;

	let referenced_by_query = r#"
		SELECT
			conname AS constraint_name,
			conrelid::regclass::text AS referencing_table,
			pg_catalog.pg_get_constraintdef(oid, true) AS constraint_def
		FROM pg_catalog.pg_constraint
		WHERE confrelid = (
			SELECT c.oid
			FROM pg_catalog.pg_class c
			LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
			WHERE n.nspname = $1 AND c.relname = $2
		)
		AND contype = 'f'
		ORDER BY conname
	"#;

	let triggers_query = r#"
		SELECT
			t.tgname AS trigger_name,
			pg_catalog.pg_get_triggerdef(t.oid, true) AS trigger_def
		FROM pg_catalog.pg_trigger t
		LEFT JOIN pg_catalog.pg_class c ON c.oid = t.tgrelid
		LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
		WHERE n.nspname = $1
			AND c.relname = $2
			AND NOT t.tgisinternal
		ORDER BY t.tgname
	"#;

	let table_info_query = r#"
		SELECT
			pg_catalog.pg_get_userbyid(c.relowner) AS owner,
			pg_size_pretty(pg_total_relation_size(c.oid)) AS size,
			CASE c.relpersistence
				WHEN 'p' THEN 'permanent'
				WHEN 'u' THEN 'unlogged'
				WHEN 't' THEN 'temporary'
			END AS persistence,
			obj_description(c.oid, 'pg_class') AS table_comment
		FROM pg_catalog.pg_class c
		LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
		WHERE n.nspname = $1 AND c.relname = $2
	"#;

	let columns_result = if sameconn {
		ctx.client
			.query(&columns_query, &[&schema, &table_name])
			.await
	} else {
		match ctx.pool.get().await {
			Ok(client) => client.query(&columns_query, &[&schema, &table_name]).await,
			Err(e) => {
				eprintln!(
					"Error getting connection from pool: {}",
					format_mobc_error(&e, None)
				);
				return ControlFlow::Continue(());
			}
		}
	};

	match columns_result {
		Ok(rows) => {
			if rows.is_empty() {
				eprintln!("No columns found for table \"{}.{}\".", schema, table_name);
				return ControlFlow::Continue(());
			}

			writer
				.writeln(&format!("Table \"{}.{}\"", schema, table_name))
				.await;

			let mut table = Table::new();
			crate::table::configure(&mut table);

			if detail {
				table.set_header(vec![
					"Column",
					"Type",
					"Collation",
					"Nullable",
					"Default",
					"Compression",
					"Storage",
				]);
			} else {
				table.set_header(vec!["Column", "Type", "Nullable", "Default"]);
			}

			for row in rows {
				let column_name: String = row.get(0);
				let data_type: String = row.get(1);
				let nullable: String = row.get(2);
				let default_value: String = row.get(3);
				let identity: String = row.get(4);
				let generated: String = row.get(5);
				let _description: Option<String> = row.get(6);
				let collation: String = row.get(7);
				let compression: String = row.get(8);
				let storage: String = row.get(9);

				let mut default_str = String::new();
				if !identity.is_empty() {
					default_str = identity;
				} else if !generated.is_empty() {
					default_str = generated;
				} else if !default_value.is_empty() {
					default_str = default_value;
				}

				if detail {
					table.add_row(vec![
						column_name,
						data_type,
						collation,
						nullable,
						default_str,
						compression,
						storage,
					]);
				} else {
					table.add_row(vec![column_name, data_type, nullable, default_str]);
				}
			}

			crate::table::style_header(&mut table);
			writer.writeln(&format!("{table}")).await;

			let indexes_result = if sameconn {
				ctx.client
					.query(indexes_query, &[&schema, &table_name])
					.await
			} else {
				match ctx.pool.get().await {
					Ok(client) => client.query(indexes_query, &[&schema, &table_name]).await,
					Err(_) => {
						return ControlFlow::Continue(());
					}
				}
			};

			if let Ok(index_rows) = indexes_result
				&& !index_rows.is_empty()
			{
				writer.writeln("\nIndexes:").await;
				for row in index_rows {
					if let (Ok(index_name), Ok(index_def), Ok(constraint_type)) = (
						row.try_get::<_, String>(0),
						row.try_get::<_, String>(1),
						row.try_get::<_, String>(2),
					) {
						if !constraint_type.is_empty() {
							writer
								.writeln(&format!(
									"    \"{index_name}\" {constraint_type} {}",
									&index_def[index_def.find("USING").unwrap_or(0)..]
								))
								.await;
						} else {
							writer
								.writeln(&format!(
									"    \"{index_name}\" {}",
									&index_def[index_def.find("USING").unwrap_or(0)..]
								))
								.await;
						}
					} else {
						writer.writeln("    Invalid index data").await;
					}
				}
			}

			let fk_result = if sameconn {
				ctx.client
					.query(foreign_keys_query, &[&schema, &table_name])
					.await
			} else {
				match ctx.pool.get().await {
					Ok(client) => {
						client
							.query(foreign_keys_query, &[&schema, &table_name])
							.await
					}
					Err(_) => {
						return ControlFlow::Continue(());
					}
				}
			};

			if let Ok(fk_rows) = fk_result
				&& !fk_rows.is_empty()
			{
				writer.writeln("\nForeign-key constraints:").await;
				for row in fk_rows {
					match (row.try_get::<_, String>(0), row.try_get::<_, String>(1)) {
						(Ok(constraint_name), Ok(constraint_def)) => {
							writer
								.writeln(&format!("    \"{}\" {}", constraint_name, constraint_def))
								.await;
						}
						err => {
							writer
								.writeln(&format!("    Invalid foreign key data: {:?}", err))
								.await;
						}
					}
				}
			}

			let check_result = if sameconn {
				ctx.client
					.query(check_constraints_query, &[&schema, &table_name])
					.await
			} else {
				match ctx.pool.get().await {
					Ok(client) => {
						client
							.query(check_constraints_query, &[&schema, &table_name])
							.await
					}
					Err(_) => {
						return ControlFlow::Continue(());
					}
				}
			};

			if let Ok(check_rows) = check_result
				&& !check_rows.is_empty()
			{
				writer.writeln("\nCheck constraints:").await;
				for row in check_rows {
					if let (Ok(constraint_name), Ok(constraint_def)) =
						(row.try_get::<_, String>(0), row.try_get::<_, String>(1))
					{
						writer
							.writeln(&format!("    \"{}\" {}", constraint_name, constraint_def))
							.await;
					} else {
						writer.writeln("    Invalid check constraint data").await;
					}
				}
			}

			if detail {
				let referenced_result = if sameconn {
					ctx.client
						.query(referenced_by_query, &[&schema, &table_name])
						.await
				} else {
					match ctx.pool.get().await {
						Ok(client) => {
							client
								.query(referenced_by_query, &[&schema, &table_name])
								.await
						}
						Err(_) => {
							return ControlFlow::Continue(());
						}
					}
				};

				if let Ok(ref_rows) = referenced_result
					&& !ref_rows.is_empty()
				{
					writer.writeln("\nReferenced by:").await;
					for row in ref_rows {
						match (
							row.try_get::<_, String>(0),
							row.try_get::<_, String>(1),
							row.try_get::<_, String>(2),
						) {
							(Ok(constraint_name), Ok(referencing_table), Ok(constraint_def)) => {
								writer
									.writeln(&format!(
										"    TABLE \"{}\" CONSTRAINT \"{}\" {}",
										referencing_table, constraint_name, constraint_def
									))
									.await;
							}
							err => {
								writer
									.writeln(&format!("    Invalid foreign key data: {:?}", err))
									.await;
							}
						}
					}
				}

				let triggers_result = if sameconn {
					ctx.client
						.query(triggers_query, &[&schema, &table_name])
						.await
				} else {
					match ctx.pool.get().await {
						Ok(client) => client.query(triggers_query, &[&schema, &table_name]).await,
						Err(_) => {
							return ControlFlow::Continue(());
						}
					}
				};

				if let Ok(trigger_rows) = triggers_result
					&& !trigger_rows.is_empty()
				{
					writer.writeln("\nTriggers:").await;
					for row in trigger_rows {
						if let Ok(trigger_def) = row.try_get::<_, String>(1) {
							writer.writeln(&format!("    {}", trigger_def)).await;
						} else {
							writer.writeln("    Invalid trigger data").await;
						}
					}
				}

				let info_result = if sameconn {
					ctx.client
						.query(table_info_query, &[&schema, &table_name])
						.await
				} else {
					match ctx.pool.get().await {
						Ok(client) => {
							client
								.query(table_info_query, &[&schema, &table_name])
								.await
						}
						Err(_) => {
							return ControlFlow::Continue(());
						}
					}
				};

				if let Ok(info_rows) = info_result
					&& let Some(row) = info_rows.first()
				{
					if let (Ok(size), Ok(persistence)) =
						(row.try_get::<_, String>(1), row.try_get::<_, String>(2))
					{
						writer.writeln(&format!("Size: {}", size)).await;
						writer
							.writeln(&format!("Persistence: {}", persistence))
							.await;
					} else {
						writer.writeln("    Invalid table info data").await;
					}
				}
			}

			writer.writeln("").await;
			ControlFlow::Continue(())
		}
		Err(e) => {
			eprintln!(
				"Error describing table \"{}.{}\": {}",
				schema,
				table_name,
				format_db_error(&e, None)
			);
			ControlFlow::Continue(())
		}
	}
}

#[cfg(test)]
mod tests {
	#[test]
	fn test_compression_column_pg13() {
		let server_version = 130000; // PostgreSQL 13
		let compression_column = if server_version >= 140000 {
			"COALESCE(a.attcompression::text, '')"
		} else {
			"''"
		};
		assert_eq!(compression_column, "''");
	}

	#[test]
	fn test_compression_column_pg14() {
		let server_version = 140000; // PostgreSQL 14
		let compression_column = if server_version >= 140000 {
			"COALESCE(a.attcompression::text, '')"
		} else {
			"''"
		};
		assert_eq!(compression_column, "COALESCE(a.attcompression::text, '')");
	}

	#[test]
	fn test_compression_column_pg15() {
		let server_version = 150000; // PostgreSQL 15
		let compression_column = if server_version >= 140000 {
			"COALESCE(a.attcompression::text, '')"
		} else {
			"''"
		};
		assert_eq!(compression_column, "COALESCE(a.attcompression::text, '')");
	}

	#[test]
	fn test_query_contains_compression_placeholder() {
		let compression_column = "TEST_PLACEHOLDER";
		let query = format!(
			r#"
			SELECT
				{} AS compression
			FROM pg_catalog.pg_class
		"#,
			compression_column
		);
		assert!(query.contains("TEST_PLACEHOLDER"));
	}
}
