1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
//! Session-level RPCs: auth, ping, SQL/DDL execution, transactions, parameters.
use nodedb_types::error::{NodeDbError, NodeDbResult};
use nodedb_types::protocol::{AuthMethod, OpCode, ResponseStatus, TextFields};
use nodedb_types::result::QueryResult;
use super::NativeConnection;
use super::response::{check_error, response_to_query_result};
impl NativeConnection {
/// Authenticate with the server.
///
/// `database` — optional target database name. When set it is sent in
/// the auth frame so the server can bind the connection's database
/// context at handshake time (equivalent to `psql -d <name>`).
pub async fn authenticate(
&mut self,
method: AuthMethod,
database: Option<&str>,
) -> NodeDbResult<()> {
let resp = self
.send(
OpCode::Auth,
TextFields {
auth: Some(method),
database: database.map(|s| s.to_string()),
..Default::default()
},
)
.await?;
if resp.status == ResponseStatus::Error {
let msg = resp
.error
.map(|e| e.message)
.unwrap_or_else(|| "auth failed".into());
return Err(NodeDbError::authorization_denied(msg));
}
self.authenticated = true;
Ok(())
}
/// Send a ping and await the pong.
pub async fn ping(&mut self) -> NodeDbResult<()> {
let resp = self.send(OpCode::Ping, TextFields::default()).await?;
if resp.status == ResponseStatus::Error {
return Err(NodeDbError::internal("ping failed"));
}
Ok(())
}
/// Execute a SQL query and return the result.
pub async fn execute_sql(&mut self, sql: &str) -> NodeDbResult<QueryResult> {
let resp = self
.send(
OpCode::Sql,
TextFields {
sql: Some(sql.to_string()),
..Default::default()
},
)
.await?;
response_to_query_result(resp)
}
/// Execute a SQL query with bound parameters and return the result.
///
/// Bound parameters travel through `TextFields::sql_params` as a
/// `Vec<Value>` — zerompk's generic array encoding serialises each
/// element via `Value`'s hand-rolled `ToMessagePack` impl so the
/// canonical scalar variants round-trip without a lossy JSON step.
/// The server inlines each value as a SQL literal before planning,
/// so `$1`, `$2`, … placeholders resolve to the caller's values.
/// Empty `params` routes through the same opcode but omits the
/// field, equivalent to `execute_sql`.
pub async fn execute_sql_with_params(
&mut self,
sql: &str,
params: &[nodedb_types::Value],
) -> NodeDbResult<QueryResult> {
let sql_params = if params.is_empty() {
None
} else {
Some(params.to_vec())
};
let resp = self
.send(
OpCode::Sql,
TextFields {
sql: Some(sql.to_string()),
sql_params,
..Default::default()
},
)
.await?;
response_to_query_result(resp)
}
/// Execute a DDL command.
pub async fn execute_ddl(&mut self, sql: &str) -> NodeDbResult<QueryResult> {
let resp = self
.send(
OpCode::Ddl,
TextFields {
sql: Some(sql.to_string()),
..Default::default()
},
)
.await?;
response_to_query_result(resp)
}
/// Begin a transaction.
pub async fn begin(&mut self) -> NodeDbResult<()> {
let resp = self.send(OpCode::Begin, TextFields::default()).await?;
check_error(&resp)
}
/// Commit the current transaction.
pub async fn commit(&mut self) -> NodeDbResult<()> {
let resp = self.send(OpCode::Commit, TextFields::default()).await?;
check_error(&resp)
}
/// Rollback the current transaction.
pub async fn rollback(&mut self) -> NodeDbResult<()> {
let resp = self.send(OpCode::Rollback, TextFields::default()).await?;
check_error(&resp)
}
/// Set a session parameter.
pub async fn set_parameter(&mut self, key: &str, value: &str) -> NodeDbResult<()> {
let resp = self
.send(
OpCode::Set,
TextFields {
key: Some(key.to_string()),
value: Some(value.to_string()),
..Default::default()
},
)
.await?;
check_error(&resp)
}
/// Show a session parameter.
pub async fn show_parameter(&mut self, key: &str) -> NodeDbResult<String> {
let resp = self
.send(
OpCode::Show,
TextFields {
key: Some(key.to_string()),
..Default::default()
},
)
.await?;
if resp.status == ResponseStatus::Error {
let msg = resp
.error
.map(|e| e.message)
.unwrap_or_else(|| "show failed".into());
return Err(NodeDbError::internal(msg));
}
let value = resp
.rows
.and_then(|rows| rows.into_iter().next())
.and_then(|row| row.into_iter().next())
.and_then(|v| v.as_str().map(String::from))
.unwrap_or_default();
Ok(value)
}
}