Invercargill-Sql/src/postgres/postgres-connection.vala

using Postgres;

namespace InvercargillSql {

    /*
     * PostgreSQL implementation of the Connection interface.
     * Wraps a libpq PGconn (Postgres.Database) for database connectivity.
     *
     * last_insert_rowid is implemented with SELECT lastval(): the last value
     * obtained from any sequence on this connection, which matches SQLite's
     * "rowid of the most recent INSERT" for identity/serial tables. If no
     * sequence has been used yet, lastval() errors and we surface 0.
     *
     * Internal class: use ConnectionFactory to create connections.
     */
    internal class PostgresConnection : Object, Connection {

        private Database? _db;

        private bool _is_open;

        private int _last_changes;

        public string conninfo { get; construct; }

        public string database_name { get; construct; }

        internal PostgresConnection(string conninfo, string database_name) {
            Object(conninfo: conninfo, database_name: database_name);
        }

        public bool is_open {
            get { return _is_open && _db != null && _db.get_status() == ConnectionStatus.OK; }
        }

        public int64 last_insert_rowid {
            get { return query_lastval(); }
        }

        public int changes {
            get { return _last_changes; }
        }

        public void open() throws SqlError {
            if (_is_open) {
                return;
            }
            _db = Postgres.connect_db(conninfo);
            if (_db.get_status() != ConnectionStatus.OK) {
                string message = _db.get_error_message();
                _db = null;
                throw new SqlError.CONNECTION_FAILED(
                    "Failed to connect to PostgreSQL database '%s': %s".printf(database_name, message)
                );
            }
            _is_open = true;
            _last_changes = 0;
        }

        public void close() throws SqlError {
            if (_is_open && _db != null) {
                _db = null;  // PQfinish via unref
                _is_open = false;
            }
        }

        public Command create_command(string sql) {
            return new PostgresCommand(this, sql);
        }

        public Transaction begin_transaction() throws SqlError {
            ensure_open();
            exec_internal("BEGIN");
            return new PostgresTransaction(this);
        }

        public void execute(string sql) throws SqlError {
            ensure_open();
            exec_internal(sql);
        }

        internal unowned Database get_database() {
            return _db;
        }

        internal void ensure_open() throws SqlError {
            if (!_is_open || _db == null) {
                throw new SqlError.CONNECTION_CLOSED("Connection is not open");
            }
        }

        /*
         * Runs SQL and checks the result status, updating the cached
         * changes counter. Shared by execute(), PostgresTransaction and
         * PostgresCommand (which keeps the owned Result for rows).
         */

        internal Postgres.Result exec_internal(string sql) throws SqlError {
            Postgres.Result result = _db.exec(sql);
            check_result(result, sql);
            update_counters(result);
            return result;
        }

        /*
         * Runs parameterized SQL via PQexecParams. n_params/values/lengths/
         * formats are the libpq arrays (values may contain nulls for SQL
         * NULL). Returns the owned PGresult after status checking.
         */

        internal Postgres.Result exec_params_internal(
            string command,
            int n_params,
            string[]? param_values,
            int[]? param_lengths,
            int[]? param_formats
        ) throws SqlError {
            Postgres.Result result = _db.exec_params(
                command, n_params, null, param_values, param_lengths, param_formats, 0
            );
            check_result(result, command);
            update_counters(result);
            return result;
        }

        /*
         * Checks a PGresult status and throws an appropriate SqlError.
         * Codes derived from SQLSTATE:
         * 08/57/58 prefixes (connection, intervention, system) -> CONNECTION_FAILED,
         * 25/26 prefixes (transaction states) -> TRANSACTION_FAILED,
         * 22xxx/42883 (data/type errors) -> TYPE_ERROR,
         * otherwise EXECUTION_FAILED.
         */

        internal void check_result(Postgres.Result result, string sql) throws SqlError {
            var status = result.get_status();
            if (status == ExecStatus.COMMAND_OK || status == ExecStatus.TUPLES_OK ||
                status == ExecStatus.EMPTY_QUERY) {
                return;
            }
            string message = result.get_error_message();
            string sqlstate = result.get_error_field(FieldCode.SQLSTATE) ?? "";
            string context = sql.length > 120 ? sql.substring(0, 120) + "..." : sql;
            string full_message = "%s (SQL: %s)".printf(message, context);
            if (sqlstate.has_prefix("08") || sqlstate.has_prefix("57") || sqlstate.has_prefix("58")) {
                _is_open = false;
                throw new SqlError.CONNECTION_FAILED(full_message);
            }
            if (sqlstate.has_prefix("25") || sqlstate.has_prefix("26")) {
                throw new SqlError.TRANSACTION_FAILED(full_message);
            }
            if (sqlstate.has_prefix("22") || sqlstate == "42883") {
                throw new SqlError.TYPE_ERROR(full_message);
            }
            throw new SqlError.EXECUTION_FAILED(full_message);
        }

        /*
         * Updates the cached changes counter from a command's PGresult
         * (PQcmdTuples). last_insert_rowid is fetched lazily via lastval().
         */

        internal void update_counters(Postgres.Result result) {
            unowned string tuples = result.get_cmd_tuples();
            int64 parsed = 0;
            if (tuples != null && tuples.length > 0 && int64.try_parse(tuples, out parsed)) {
                _last_changes = (int)parsed;
            } else {
                _last_changes = 0;
            }
        }

        /*
         * Runs SELECT lastval() on this connection.
         * Returns 0 when no sequence has been used yet in the session
         * (lastval() raises 55006 then, which is not fatal for callers).
         */

        private int64 query_lastval() throws SqlError {
            ensure_open();
            Postgres.Result result = _db.exec("SELECT lastval()");
            var status = result.get_status();
            if (status == ExecStatus.TUPLES_OK && result.get_n_tuples() > 0) {
                return int64.parse(result.get_value(0, 0));
            }
            return 0;
        }
    }
}