local mysql = {}

local Connection = {}
Connection.__index = Connection

function Connection:close()
	if self._id then
		local ok = moonshark.sql_close(self._id)
		self._id = nil
		return ok
	end
	return false
end

function Connection:ping()
	if not self._id then
		error("Connection is closed")
	end
	return moonshark.sql_ping(self._id)
end

function Connection:query(query_str, ...)
	if not self._id then
		error("Connection is closed")
	end
	return moonshark.sql_query(self._id, query_str:normalize_whitespace(), ...)
end

function Connection:exec(query_str, ...)
	if not self._id then
		error("Connection is closed")
	end
	return moonshark.sql_exec(self._id, query_str:normalize_whitespace(), ...)
end

function Connection:query_row(query_str, ...)
	local results = self:query(query_str, ...)
	return results and #results > 0 and results[1] or nil
end

function Connection:query_value(query_str, ...)
	local row = self:query_row(query_str, ...)
	if row then
		for _, value in pairs(row) do
			return value
		end
	end
	return nil
end

function Connection:begin()
	local result = self:exec("BEGIN")
	if result then
		return {
			conn = self,
			active = true,
			commit = function(tx)
				if tx.active then
					tx.active = false
					return tx.conn:exec("COMMIT")
				end
				return false
			end,
			rollback = function(tx)
				if tx.active then
					tx.active = false
					return tx.conn:exec("ROLLBACK")
				end
				return false
			end,
			savepoint = function(tx, name)
				if not tx.active then error("Transaction is not active") end
				if name:is_blank() then error("Savepoint name cannot be empty") end
				return tx.conn:exec("SAVEPOINT {{name}}":parse({name = name}))
			end,
			rollback_to = function(tx, name)
				if not tx.active then error("Transaction is not active") end
				if name:is_blank() then error("Savepoint name cannot be empty") end
				return tx.conn:exec("ROLLBACK TO SAVEPOINT {{name}}":parse({name = name}))
			end,
			query = function(tx, query_str, ...)
				if not tx.active then error("Transaction is not active") end
				return tx.conn:query(query_str, ...)
			end,
			exec = function(tx, query_str, ...)
				if not tx.active then error("Transaction is not active") end
				return tx.conn:exec(query_str, ...)
			end,
			query_row = function(tx, query_str, ...)
				if not tx.active then error("Transaction is not active") end
				return tx.conn:query_row(query_str, ...)
			end,
			query_value = function(tx, query_str, ...)
				if not tx.active then error("Transaction is not active") end
				return tx.conn:query_value(query_str, ...)
			end
		}
	end
	return nil
end

function Connection:insert(table_name, data)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end

	local keys = table.keys(data)
	local values = table.values(data)
	local placeholders = string.repeat_("?, ", #keys):trim_right(", ")

	local query = "INSERT INTO {{table}} ({{columns}}) VALUES ({{placeholders}})":parse({
		table = table_name,
		columns = keys:join(", "),
		placeholders = placeholders
	})

	return self:exec(query, unpack(values))
end

function Connection:upsert(table_name, data, update_data)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end

	local keys = table.keys(data)
	local values = table.values(data)
	local placeholders = string.repeat_("?, ", #keys):trim_right(", ")

	-- Use update_data if provided, otherwise update with same data
	local update_source = update_data or data
	local updates = table.map(table.keys(update_source), function(key)
		return key .. " = VALUES(" .. key .. ")"
	end)

	local query = "INSERT INTO {{table}} ({{columns}}) VALUES ({{placeholders}}) ON DUPLICATE KEY UPDATE {{updates}}":parse({
		table = table_name,
		columns = keys:join(", "),
		placeholders = placeholders,
		updates = updates:join(", ")
	})

	return self:exec(query, unpack(values))
end

function Connection:replace(table_name, data)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end

	local keys = table.keys(data)
	local values = table.values(data)
	local placeholders = string.repeat_("?, ", #keys):trim_right(", ")

	local query = "REPLACE INTO {{table}} ({{columns}}) VALUES ({{placeholders}})":parse({
		table = table_name,
		columns = keys:join(", "),
		placeholders = placeholders
	})

	return self:exec(query, unpack(values))
end

function Connection:update(table_name, data, where_clause, ...)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end
	if where_clause:is_blank() then
		error("WHERE clause cannot be empty for UPDATE")
	end

	local keys = table.keys(data)
	local values = table.values(data)
	local sets = table.map(keys, function(key) return key .. " = ?" end)

	local query = "UPDATE {{table}} SET {{sets}} WHERE {{where}}":parse({
		table = table_name,
		sets = sets:join(", "),
		where = where_clause
	})

	table.extend(values, {...})
	return self:exec(query, unpack(values))
end

function Connection:delete(table_name, where_clause, ...)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end
	if where_clause:is_blank() then
		error("WHERE clause cannot be empty for DELETE")
	end

	local query = "DELETE FROM {{table}} WHERE {{where}}":parse({
		table = table_name,
		where = where_clause
	})
	return self:exec(query, ...)
end

function Connection:select(table_name, columns, where_clause, ...)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end

	columns = columns or "*"
	if type(columns) == "table" then
		columns = table.concat(columns, ", ")
	end

	if where_clause and not where_clause:is_blank() then
		local query = "SELECT {{columns}} FROM {{table}} WHERE {{where}}":parse({
			columns = columns,
			table = table_name,
			where = where_clause
		})
		return self:query(query, ...)
	else
		local query = "SELECT {{columns}} FROM {{table}}":parse({
			columns = columns,
			table = table_name
		})
		return self:query(query)
	end
end

-- MySQL schema helpers
function Connection:database_exists(database_name)
	if database_name:is_blank() then return false end
	return self:query_value("SELECT SCHEMA_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = ?",
		database_name:trim()) ~= nil
end

function Connection:table_exists(table_name, database_name)
	if table_name:is_blank() then return false end
	database_name = database_name or self:current_database()
	if not database_name then return false end

	return self:query_value("SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?",
		database_name:trim(), table_name:trim()) ~= nil
end

function Connection:column_exists(table_name, column_name, database_name)
	if table_name:is_blank() or column_name:is_blank() then return false end
	database_name = database_name or self:current_database()
	if not database_name then return false end

	return self:query_value([[
		SELECT COLUMN_NAME FROM information_schema.COLUMNS
		WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND COLUMN_NAME = ?
	]], database_name:trim(), table_name:trim(), column_name:trim()) ~= nil
end

function Connection:create_database(database_name, charset, collation)
	if database_name:is_blank() then
		error("Database name cannot be empty")
	end

	local charset_clause = charset and " CHARACTER SET " .. charset or ""
	local collation_clause = collation and " COLLATE " .. collation or ""

	return self:exec("CREATE DATABASE IF NOT EXISTS {{database}}{{charset}}{{collation}}":parse({
		database = database_name,
		charset = charset_clause,
		collation = collation_clause
	}))
end

function Connection:drop_database(database_name)
	if database_name:is_blank() then
		error("Database name cannot be empty")
	end
	return self:exec("DROP DATABASE IF EXISTS {{database}}":parse({database = database_name}))
end

function Connection:create_table(table_name, schema, engine, charset)
	if table_name:is_blank() or schema:is_blank() then
		error("Table name and schema cannot be empty")
	end

	local engine_clause = engine and " ENGINE=" .. engine:upper() or ""
	local charset_clause = charset and " CHARACTER SET " .. charset or ""

	return self:exec("CREATE TABLE IF NOT EXISTS {{table}} ({{schema}}){{engine}}{{charset}}":parse({
		table = table_name,
		schema = schema:trim(),
		engine = engine_clause,
		charset = charset_clause
	}))
end

function Connection:drop_table(table_name)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end
	return self:exec("DROP TABLE IF EXISTS {{table}}":parse({table = table_name}))
end

function Connection:add_column(table_name, column_def, position)
	if table_name:is_blank() or column_def:is_blank() then
		error("Table name and column definition cannot be empty")
	end

	local position_clause = position and " " .. position or ""
	return self:exec("ALTER TABLE {{table}} ADD COLUMN {{column}}{{position}}":parse({
		table = table_name,
		column = column_def:trim(),
		position = position_clause
	}))
end

function Connection:drop_column(table_name, column_name)
	if table_name:is_blank() or column_name:is_blank() then
		error("Table name and column name cannot be empty")
	end
	return self:exec("ALTER TABLE {{table}} DROP COLUMN {{column}}":parse({
		table = table_name,
		column = column_name
	}))
end

function Connection:modify_column(table_name, column_def)
	if table_name:is_blank() or column_def:is_blank() then
		error("Table name and column definition cannot be empty")
	end
	return self:exec("ALTER TABLE {{table}} MODIFY COLUMN {{column}}":parse({
		table = table_name,
		column = column_def:trim()
	}))
end

function Connection:rename_table(old_name, new_name)
	if old_name:is_blank() or new_name:is_blank() then
		error("Old and new table names cannot be empty")
	end
	return self:exec("RENAME TABLE {{old}} TO {{new}}":parse({old = old_name, new = new_name}))
end

function Connection:create_index(index_name, table_name, columns, unique, type)
	if index_name:is_blank() or table_name:is_blank() then
		error("Index name and table name cannot be empty")
	end

	local unique_clause = unique and "UNIQUE " or ""
	local type_clause = type and " USING " .. type:upper() or ""
	local columns_str = type(columns) == "table" and table.concat(columns, ", ") or tostring(columns)

	return self:exec("CREATE {{unique}}INDEX {{index}} ON {{table}} ({{columns}}){{type}}":parse({
		unique = unique_clause,
		index = index_name,
		table = table_name,
		columns = columns_str,
		type = type_clause
	}))
end

function Connection:drop_index(index_name, table_name)
	if index_name:is_blank() or table_name:is_blank() then
		error("Index name and table name cannot be empty")
	end
	return self:exec("DROP INDEX {{index}} ON {{table}}":parse({
		index = index_name,
		table = table_name
	}))
end

-- MySQL maintenance functions
function Connection:optimize(table_name)
	local table_clause = table_name and " " .. table_name or ""
	return self:query("OPTIMIZE TABLE{{table}}":parse({table = table_clause}))
end

function Connection:repair(table_name)
	if table_name:is_blank() then
		error("Table name cannot be empty for REPAIR")
	end
	return self:query("REPAIR TABLE {{table}}":parse({table = table_name}))
end

function Connection:check_table(table_name, options)
	if table_name:is_blank() then
		error("Table name cannot be empty for CHECK")
	end

	local options_clause = ""
	if options then
		local valid_options = {"QUICK", "FAST", "MEDIUM", "EXTENDED", "CHANGED"}
		local options_upper = options:upper()

		if table.contains(valid_options, options_upper) then
			options_clause = " " .. options_upper
		end
	end

	return self:query("CHECK TABLE {{table}}{{options}}":parse({
		table = table_name,
		options = options_clause
	}))
end

function Connection:analyze_table(table_name)
	if table_name:is_blank() then
		error("Table name cannot be empty for ANALYZE")
	end
	return self:query("ANALYZE TABLE {{table}}":parse({table = table_name}))
end

-- MySQL settings and introspection
function Connection:show(what)
	if what:is_blank() then
		error("SHOW parameter cannot be empty")
	end
	return self:query("SHOW {{what}}":parse({what = what:upper()}))
end

function Connection:current_database()
	return self:query_value("SELECT DATABASE() AS db")
end

function Connection:version()
	return self:query_value("SELECT VERSION() AS version")
end

function Connection:connection_id()
	return self:query_value("SELECT CONNECTION_ID()")
end

function Connection:list_databases()
	return self:query("SHOW DATABASES")
end

function Connection:list_tables(database_name)
	if database_name and not database_name:is_blank() then
		return self:query("SHOW TABLES FROM {{database}}":parse({database = database_name}))
	else
		return self:query("SHOW TABLES")
	end
end

function Connection:describe_table(table_name)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end
	return self:query("DESCRIBE {{table}}":parse({table = table_name}))
end

function Connection:show_create_table(table_name)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end
	return self:query("SHOW CREATE TABLE {{table}}":parse({table = table_name}))
end

function Connection:show_indexes(table_name)
	if table_name:is_blank() then
		error("Table name cannot be empty")
	end
	return self:query("SHOW INDEXES FROM {{table}}":parse({table = table_name}))
end

function Connection:show_table_status(table_name)
	if table_name and not table_name:is_blank() then
		return self:query("SHOW TABLE STATUS LIKE ?", table_name)
	else
		return self:query("SHOW TABLE STATUS")
	end
end

-- MySQL user and privilege management
function Connection:create_user(username, password, host)
	if username:is_blank() or password:is_blank() then
		error("Username and password cannot be empty")
	end

	host = host or "%"
	return self:exec("CREATE USER '{{username}}'@'{{host}}' IDENTIFIED BY ?":parse({
		username = username,
		host = host
	}), password)
end

function Connection:drop_user(username, host)
	if username:is_blank() then
		error("Username cannot be empty")
	end

	host = host or "%"
	return self:exec("DROP USER IF EXISTS '{{username}}'@'{{host}}'":parse({
		username = username,
		host = host
	}))
end

function Connection:grant(privileges, database, table_name, username, host)
	if privileges:is_blank() or database:is_blank() or username:is_blank() then
		error("Privileges, database, and username cannot be empty")
	end

	host = host or "%"
	table_name = table_name or "*"
	local object = database .. "." .. table_name

	return self:exec("GRANT {{privileges}} ON {{object}} TO '{{username}}'@'{{host}}'":parse({
		privileges = privileges:upper(),
		object = object,
		username = username,
		host = host
	}))
end

function Connection:revoke(privileges, database, table_name, username, host)
	if privileges:is_blank() or database:is_blank() or username:is_blank() then
		error("Privileges, database, and username cannot be empty")
	end

	host = host or "%"
	table_name = table_name or "*"
	local object = database .. "." .. table_name

	return self:exec("REVOKE {{privileges}} ON {{object}} FROM '{{username}}'@'{{host}}'":parse({
		privileges = privileges:upper(),
		object = object,
		username = username,
		host = host
	}))
end

function Connection:flush_privileges()
	return self:exec("FLUSH PRIVILEGES")
end

-- MySQL variables and configuration
function Connection:set_variable(name, value, global)
	if name:is_blank() then
		error("Variable name cannot be empty")
	end

	local scope = global and "GLOBAL " or "SESSION "
	return self:exec("SET {{scope}}{{name}} = ?":parse({scope = scope, name = name}), value)
end

function Connection:get_variable(name, global)
	if name:is_blank() then
		error("Variable name cannot be empty")
	end

	local scope = global and "global." or "session."
	return self:query_value("SELECT @@{{scope}}{{name}}":parse({scope = scope, name = name}))
end

function Connection:show_variables(pattern)
	if pattern and not pattern:is_blank() then
		return self:query("SHOW VARIABLES LIKE ?", pattern)
	else
		return self:query("SHOW VARIABLES")
	end
end

function Connection:show_status(pattern)
	if pattern and not pattern:is_blank() then
		return self:query("SHOW STATUS LIKE ?", pattern)
	else
		return self:query("SHOW STATUS")
	end
end

-- Connection management
function mysql.connect(dsn)
	if dsn:is_blank() then
		error("DSN cannot be empty")
	end

	local conn_id = moonshark.sql_connect("mysql", dsn:trim())
	if conn_id then
		return setmetatable({_id = conn_id}, Connection)
	end
	return nil
end

mysql.open = mysql.connect

-- Quick execution functions
function mysql.query(dsn, query_str, ...)
	local conn = mysql.connect(dsn)
	if not conn then
		error("Failed to connect to MySQL database")
	end

	local results = conn:query(query_str, ...)
	conn:close()
	return results
end

function mysql.exec(dsn, query_str, ...)
	local conn = mysql.connect(dsn)
	if not conn then
		error("Failed to connect to MySQL database")
	end

	local result = conn:exec(query_str, ...)
	conn:close()
	return result
end

function mysql.query_row(dsn, query_str, ...)
	local results = mysql.query(dsn, query_str, ...)
	return results and #results > 0 and results[1] or nil
end

function mysql.query_value(dsn, query_str, ...)
	local row = mysql.query_row(dsn, query_str, ...)
	if row then
		for _, value in pairs(row) do
			return value
		end
	end
	return nil
end

-- Migration helpers
function mysql.migrate(dsn, migrations, database_name)
	local conn = mysql.connect(dsn)
	if not conn then
		error("Failed to connect to MySQL database for migration")
	end

	-- Use specified database if provided
	if database_name and not database_name:is_blank() then
		conn:exec("USE {{database}}":parse({database = database_name}))
	end

	-- Create migrations table
	conn:create_table("_migrations", "id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255) UNIQUE NOT NULL, applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP")

	local tx = conn:begin()
	if not tx then
		conn:close()
		error("Failed to begin migration transaction")
	end

	for _, migration in ipairs(migrations) do
		if not migration.name or migration.name:is_blank() then
			tx:rollback()
			conn:close()
			error("Migration must have a non-empty name")
		end

		-- Check if migration already applied
		local existing = conn:query_value("SELECT id FROM _migrations WHERE name = ?", migration.name:trim())
		if not existing then
			local ok, err = pcall(function()
				if type(migration.up) == "string" then
					conn:exec(migration.up)
				elseif type(migration.up) == "function" then
					migration.up(conn)
				else
					error("Migration 'up' must be string or function")
				end
			end)

			if ok then
				conn:exec("INSERT INTO _migrations (name) VALUES (?)", migration.name:trim())
				print("Applied migration: {{name}}":parse({name = migration.name}))
			else
				tx:rollback()
				conn:close()
				error("Migration '{{name}}' failed: {{error}}":parse({
					name = migration.name,
					error = err or "unknown error"
				}))
			end
		end
	end

	tx:commit()
	conn:close()
	return true
end

-- Result processing utilities
function mysql.to_array(results, column_name)
	if not results or table.is_empty(results) then return {} end
	if column_name:is_blank() then error("Column name cannot be empty") end
	return table.map(results, function(row) return row[column_name] end)
end

function mysql.to_map(results, key_column, value_column)
	if not results or table.is_empty(results) then return {} end
	if key_column:is_blank() then error("Key column name cannot be empty") end

	local map = {}
	for _, row in ipairs(results) do
		local key = row[key_column]
		map[key] = value_column and row[value_column] or row
	end
	return map
end

function mysql.group_by(results, column_name)
	if not results or table.is_empty(results) then return {} end
	if column_name:is_blank() then error("Column name cannot be empty") end
	return table.group_by(results, function(row) return row[column_name] end)
end

function mysql.print_results(results)
	if not results or table.is_empty(results) then
		print("No results")
		return
	end

	local columns = table.keys(results[1])
	table.sort(columns)

	-- Calculate column widths
	local widths = {}
	for _, col in ipairs(columns) do
		widths[col] = col:length()
		for _, row in ipairs(results) do
			local value = tostring(row[col] or "")
			widths[col] = math.max(widths[col], value:length())
		end
	end

	-- Print header and separator
	local header_parts = table.map(columns, function(col) return col:pad_right(widths[col]) end)
	local separator_parts = table.map(columns, function(col) return string.repeat_("-", widths[col]) end)

	print(table.concat(header_parts, " | "))
	print(table.concat(separator_parts, "-+-"))

	-- Print rows
	for _, row in ipairs(results) do
		local value_parts = table.map(columns, function(col)
			local value = tostring(row[col] or "")
			return value:pad_right(widths[col])
		end)
		print(table.concat(value_parts, " | "))
	end
end

-- MySQL-specific utilities
function mysql.escape_string(str_val)
	if type(str_val) ~= "string" then
		return tostring(str_val)
	end
	return str_val:replace("'", "\\'")
end

function mysql.escape_identifier(name)
	if name:is_blank() then
		error("Identifier name cannot be empty")
	end
	return "`{{name}}`":parse({name = name:replace("`", "``")})
end

-- DSN builder helper
function mysql.build_dsn(options)
	if type(options) ~= "table" then
		error("Options must be a table")
	end

	local parts = {}

	if options.username and not options.username:is_blank() then
		table.insert(parts, options.username)
		if options.password and not options.password:is_blank() then
			parts[#parts] = parts[#parts] .. ":" .. options.password
		end
		parts[#parts] = parts[#parts] .. "@"
	end

	if options.protocol and not options.protocol:is_blank() then
		local host_part = options.protocol .. "("
		if options.host and not options.host:is_blank() then
			host_part = host_part .. options.host
			if options.port then
				host_part = host_part .. ":" .. tostring(options.port)
			end
		end
		table.insert(parts, host_part .. ")")
	elseif options.host and not options.host:is_blank() then
		local host_part = "tcp(" .. options.host
		if options.port then
			host_part = host_part .. ":" .. tostring(options.port)
		end
		table.insert(parts, host_part .. ")")
	end

	if options.database and not options.database:is_blank() then
		table.insert(parts, "/" .. options.database)
	end

	-- Add parameters
	local params = {}
	if options.charset and not options.charset:is_blank() then
		table.insert(params, "charset=" .. options.charset)
	end
	if options.parseTime ~= nil then
		table.insert(params, "parseTime=" .. tostring(options.parseTime))
	end
	if options.timeout and not options.timeout:is_blank() then
		table.insert(params, "timeout=" .. options.timeout)
	end
	if options.tls and not options.tls:is_blank() then
		table.insert(params, "tls=" .. options.tls)
	end

	if #params > 0 then
		table.insert(parts, "?" .. table.concat(params, "&"))
	end

	return table.concat(parts, "")
end

return mysql