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
174
175
176
177
178
179
180
181
182
183
184
185
186
// SPDX-License-Identifier: BUSL-1.1
//! Protocol-neutral atomic transfer SQL functions: TRANSFER (fungible) and
//! TRANSFER_ITEM (non-fungible).
//!
//! `SELECT TRANSFER(collection, source_key, dest_key, field, amount)`
//! — Atomically: source.field -= amount, dest.field += amount.
//! — Fails with INSUFFICIENT_BALANCE if source.field < amount.
//! — Returns: `{ source_key, dest_key, field, amount, source_balance, dest_balance }`.
//!
//! `SELECT TRANSFER_ITEM(source_collection, dest_collection, item_id, source_owner, dest_owner)`
//! — Atomically: remove item from source owner, add to dest owner.
//! — Fails with NOT_FOUND if source doesn't own the item.
//! — Returns: `{ item_key, dest_key, source_collection, dest_collection }`.
//!
//! Both dispatch to the Data Plane as dedicated KvOp variants. The entire
//! read-validate-write executes in a single TPC core pass — no TOCTOU race.
use crate::control::security::identity::AuthenticatedIdentity;
use crate::control::server::shared::session::DmlTxnCtx;
use crate::control::state::SharedState;
use crate::types::{DatabaseId, VShardId};
use nodedb_physical::physical_plan::{KvOp, PhysicalPlan};
use super::super::result::{DdlError, DdlResult};
use super::kv_atomic::{dispatch_and_respond, parse_function_args};
/// Handle `SELECT TRANSFER(collection, source_key, dest_key, field, amount)`
pub async fn transfer(
state: &SharedState,
identity: &AuthenticatedIdentity,
sql: &str,
txn_ctx: &DmlTxnCtx<'_>,
) -> Result<Vec<DdlResult>, DdlError> {
let args = parse_function_args(sql, "TRANSFER")?;
if args.len() < 5 {
return Err(ddl_err(
"42601",
"TRANSFER requires 5 arguments: (collection, source_key, dest_key, field, amount)",
));
}
let collection = unquote(&args[0]).to_lowercase();
let source_key = unquote(&args[1]);
let dest_key = unquote(&args[2]);
let field = unquote(&args[3]);
let amount_str = args[4].trim().to_string();
let amount: f64 = amount_str.parse().map_err(|_| {
ddl_err(
"42601",
format!("TRANSFER: amount must be a number, got '{amount_str}'"),
)
})?;
if amount <= 0.0 {
return Err(ddl_err("42601", "TRANSFER: amount must be positive"));
}
let vshard = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &collection);
// Dispatch to Data Plane — entire read+validate+write is atomic (single TPC
// core). Routed through the protocol-neutral in-transaction staging gate
// (`dispatch_and_respond`, shared with `KV_INCR` et al.): outside a
// transaction it dispatches immediately, byte-identical to before;
// inside a `BEGIN..COMMIT` block `KvOp::Transfer` is staged into the
// per-transaction overlay so a same-transaction read observes both
// updated balances and COMMIT durably replays the same op.
// Content-addressed cross-engine identity per key: the debited (source)
// row and the credited (dest) row each keep the surrogate their original
// insert assigned. Distinct keys → distinct surrogates, so the two rows
// never collapse onto one identity.
let source_bytes = source_key.into_bytes();
let dest_bytes = dest_key.into_bytes();
let debit_surrogate = state
.surrogate_assigner
.assign(
DatabaseId::DEFAULT,
identity.tenant_id,
&collection,
&source_bytes,
)
.map_err(|e| ddl_err("XX000", e.to_string()))?;
let credit_surrogate = state
.surrogate_assigner
.assign(
DatabaseId::DEFAULT,
identity.tenant_id,
&collection,
&dest_bytes,
)
.map_err(|e| ddl_err("XX000", e.to_string()))?;
let plan = PhysicalPlan::Kv(KvOp::Transfer {
collection,
source_key: source_bytes,
dest_key: dest_bytes,
field,
amount,
debit_surrogate,
credit_surrogate,
});
dispatch_and_respond(state, identity, vshard, plan, "TRANSFER", txn_ctx).await
}
/// Handle `SELECT TRANSFER_ITEM(source_collection, dest_collection, item_id, source_owner, dest_owner)`
pub async fn transfer_item(
state: &SharedState,
identity: &AuthenticatedIdentity,
sql: &str,
txn_ctx: &DmlTxnCtx<'_>,
) -> Result<Vec<DdlResult>, DdlError> {
let args = parse_function_args(sql, "TRANSFER_ITEM")?;
if args.len() < 5 {
return Err(ddl_err(
"42601",
"TRANSFER_ITEM requires 5 arguments: (source_collection, dest_collection, item_id, source_owner, dest_owner)",
));
}
let source_collection = unquote(&args[0]).to_lowercase();
let dest_collection = unquote(&args[1]).to_lowercase();
let item_id = unquote(&args[2]);
let source_owner = unquote(&args[3]);
let dest_owner = unquote(&args[4]);
// Cross-collection transfers must be on the same vshard.
// Validate this upfront to prevent silent failures.
let vshard_src = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &source_collection);
let vshard_dst = VShardId::from_collection_in_database(DatabaseId::DEFAULT, &dest_collection);
if source_collection != dest_collection && vshard_src != vshard_dst {
return Err(ddl_err(
"0A000",
format!(
"TRANSFER_ITEM: cross-shard transfer not supported \
(source '{}' and dest '{}' map to different vShards)",
source_collection, dest_collection
),
));
}
let item_key = format!("{source_owner}:{item_id}");
let dest_key = format!("{dest_owner}:{item_id}");
// The moved row's identity is content-addressed at its DESTINATION
// `(dest_collection, dest_key)`, matching the engine write-back.
let dest_bytes = dest_key.into_bytes();
let surrogate = state
.surrogate_assigner
.assign(
DatabaseId::DEFAULT,
identity.tenant_id,
&dest_collection,
&dest_bytes,
)
.map_err(|e| ddl_err("XX000", e.to_string()))?;
// Dispatch to Data Plane — verify + delete + insert is atomic. Routed
// through the same in-transaction staging gate as `TRANSFER` (see above).
let plan = PhysicalPlan::Kv(KvOp::TransferItem {
source_collection,
dest_collection,
item_key: item_key.into_bytes(),
dest_key: dest_bytes,
surrogate,
});
dispatch_and_respond(state, identity, vshard_src, plan, "TRANSFER_ITEM", txn_ctx).await
}
// ── Helpers ────────────────────────────────────────────────────────────
fn unquote(s: &str) -> String {
let t = s.trim();
if t.starts_with('\'') && t.ends_with('\'') && t.len() >= 2 {
t[1..t.len() - 1].to_string()
} else {
t.to_string()
}
}
fn ddl_err(sqlstate: &str, message: impl Into<String>) -> DdlError {
DdlError {
sqlstate: sqlstate.to_string(),
message: message.into(),
}
}