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; } } }