1use crate::WorkerError;
2use crate::api;
3use crate::error::Error;
4use serde::{Deserialize, Deserializer, Serialize, Serializer};
5
6fn call_broker<Request, Response>(
7 operation: &str,
8 request: &Request,
9) -> Result<Response, WorkerError>
10where
11 Request: Serialize,
12 Response: for<'de> Deserialize<'de>,
13{
14 api::call(operation, request).map_err(worker_error)
15}
16
17fn worker_error(error: Error) -> WorkerError {
18 error.into()
19}
20
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22#[serde(deny_unknown_fields)]
23pub struct Usage {
24 pub plugin_instance_id: String,
25 pub store_id: String,
26 pub usage_bytes: i64,
27 pub quota_bytes: i64,
28 pub usage_files: i64,
29 pub quota_files: i64,
30}
31
32#[derive(Serialize)]
33struct OperationRequest<'a, Request> {
34 operation: &'static str,
35 #[serde(flatten)]
36 request: &'a Request,
37}
38
39pub mod storage {
40 use super::*;
41
42 pub mod files {
43 use super::*;
44
45 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46 #[serde(deny_unknown_fields)]
47 pub struct ReadRequest {
48 pub store_id: String,
49 pub path: String,
50 #[serde(default, skip_serializing_if = "Option::is_none")]
51 pub max_bytes: Option<u64>,
52 }
53
54 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
55 #[serde(deny_unknown_fields)]
56 pub struct ReadResponse {
57 pub ok: bool,
58 pub path: String,
59 pub data_base64: String,
60 pub size_bytes: i64,
61 pub usage: Usage,
62 }
63
64 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65 #[serde(deny_unknown_fields)]
66 pub struct WriteRequest {
67 pub store_id: String,
68 pub path: String,
69 pub data_base64: String,
70 }
71
72 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
73 #[serde(deny_unknown_fields)]
74 pub struct WriteResponse {
75 pub ok: bool,
76 pub path: String,
77 pub size_bytes: i64,
78 pub usage: Usage,
79 }
80
81 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
82 #[serde(deny_unknown_fields)]
83 pub struct DeleteRequest {
84 pub store_id: String,
85 pub path: String,
86 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
87 pub recursive: bool,
88 }
89
90 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
91 #[serde(deny_unknown_fields)]
92 pub struct DeleteResponse {
93 pub ok: bool,
94 pub path: String,
95 }
96
97 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98 #[serde(deny_unknown_fields)]
99 pub struct ListRequest {
100 pub store_id: String,
101 #[serde(default, skip_serializing_if = "String::is_empty")]
102 pub path: String,
103 #[serde(default, skip_serializing_if = "Option::is_none")]
104 pub max_entries: Option<u32>,
105 }
106
107 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
108 #[serde(deny_unknown_fields)]
109 pub struct Entry {
110 pub path: String,
111 pub dir: bool,
112 #[serde(default)]
113 pub size_bytes: i64,
114 pub updated_at: String,
115 }
116
117 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
118 #[serde(deny_unknown_fields)]
119 pub struct ListResponse {
120 pub ok: bool,
121 pub path: String,
122 pub entries: Vec<Entry>,
123 pub usage: Usage,
124 }
125
126 pub fn read(request: ReadRequest) -> Result<ReadResponse, WorkerError> {
127 call_broker(
128 "storage.files",
129 &OperationRequest {
130 operation: "read",
131 request: &request,
132 },
133 )
134 }
135
136 pub fn write(request: WriteRequest) -> Result<WriteResponse, WorkerError> {
137 call_broker(
138 "storage.files",
139 &OperationRequest {
140 operation: "write",
141 request: &request,
142 },
143 )
144 }
145
146 pub fn delete(request: DeleteRequest) -> Result<DeleteResponse, WorkerError> {
147 call_broker(
148 "storage.files",
149 &OperationRequest {
150 operation: "delete",
151 request: &request,
152 },
153 )
154 }
155
156 pub fn list(request: ListRequest) -> Result<ListResponse, WorkerError> {
157 call_broker(
158 "storage.files",
159 &OperationRequest {
160 operation: "list",
161 request: &request,
162 },
163 )
164 }
165 }
166
167 pub mod kv {
168 use super::*;
169
170 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171 #[serde(deny_unknown_fields)]
172 pub struct GetRequest {
173 pub store_id: String,
174 pub key: String,
175 #[serde(default, skip_serializing_if = "Option::is_none")]
176 pub max_bytes: Option<u64>,
177 }
178
179 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
180 #[serde(deny_unknown_fields)]
181 pub struct GetResponse {
182 pub ok: bool,
183 pub key: String,
184 pub value_base64: String,
185 pub size_bytes: i64,
186 pub usage: Usage,
187 }
188
189 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
190 #[serde(deny_unknown_fields)]
191 pub struct PutRequest {
192 pub store_id: String,
193 pub key: String,
194 pub value_base64: String,
195 }
196
197 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
198 #[serde(deny_unknown_fields)]
199 pub struct PutResponse {
200 pub ok: bool,
201 pub key: String,
202 pub size_bytes: i64,
203 pub usage: Usage,
204 }
205
206 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
207 #[serde(deny_unknown_fields)]
208 pub struct DeleteRequest {
209 pub store_id: String,
210 pub key: String,
211 }
212
213 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
214 #[serde(deny_unknown_fields)]
215 pub struct DeleteResponse {
216 pub ok: bool,
217 pub key: String,
218 }
219
220 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
221 #[serde(deny_unknown_fields)]
222 pub struct ListRequest {
223 pub store_id: String,
224 #[serde(default, skip_serializing_if = "String::is_empty")]
225 pub prefix: String,
226 #[serde(default, skip_serializing_if = "Option::is_none")]
227 pub max_entries: Option<u32>,
228 }
229
230 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
231 #[serde(deny_unknown_fields)]
232 pub struct Entry {
233 pub key: String,
234 pub size_bytes: i64,
235 pub updated_at: String,
236 }
237
238 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
239 #[serde(deny_unknown_fields)]
240 pub struct ListResponse {
241 pub ok: bool,
242 #[serde(default)]
243 pub prefix: String,
244 pub entries: Vec<Entry>,
245 pub usage: Usage,
246 }
247
248 pub fn get(request: GetRequest) -> Result<GetResponse, WorkerError> {
249 call_broker(
250 "storage.kv",
251 &OperationRequest {
252 operation: "get",
253 request: &request,
254 },
255 )
256 }
257
258 pub fn put(request: PutRequest) -> Result<PutResponse, WorkerError> {
259 call_broker(
260 "storage.kv",
261 &OperationRequest {
262 operation: "put",
263 request: &request,
264 },
265 )
266 }
267
268 pub fn delete(request: DeleteRequest) -> Result<DeleteResponse, WorkerError> {
269 call_broker(
270 "storage.kv",
271 &OperationRequest {
272 operation: "delete",
273 request: &request,
274 },
275 )
276 }
277
278 pub fn list(request: ListRequest) -> Result<ListResponse, WorkerError> {
279 call_broker(
280 "storage.kv",
281 &OperationRequest {
282 operation: "list",
283 request: &request,
284 },
285 )
286 }
287 }
288
289 pub mod sqlite {
290 use super::*;
291
292 #[derive(Debug, Clone, PartialEq)]
293 pub enum Value {
294 Null,
295 Integer(i64),
296 Float(f64),
297 Text(String),
298 BlobBase64(String),
299 }
300
301 #[derive(Serialize, Deserialize)]
302 #[serde(deny_unknown_fields)]
303 struct ValueWire {
304 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
305 null: bool,
306 #[serde(default, skip_serializing_if = "Option::is_none")]
307 int: Option<i64>,
308 #[serde(default, skip_serializing_if = "Option::is_none")]
309 float: Option<f64>,
310 #[serde(default, skip_serializing_if = "Option::is_none")]
311 text: Option<String>,
312 #[serde(default, skip_serializing_if = "Option::is_none")]
313 blob_base64: Option<String>,
314 }
315
316 impl Serialize for Value {
317 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
318 where
319 S: Serializer,
320 {
321 let wire = match self {
322 Self::Null => ValueWire {
323 null: true,
324 int: None,
325 float: None,
326 text: None,
327 blob_base64: None,
328 },
329 Self::Integer(value) => ValueWire {
330 null: false,
331 int: Some(*value),
332 float: None,
333 text: None,
334 blob_base64: None,
335 },
336 Self::Float(value) => ValueWire {
337 null: false,
338 int: None,
339 float: Some(*value),
340 text: None,
341 blob_base64: None,
342 },
343 Self::Text(value) => ValueWire {
344 null: false,
345 int: None,
346 float: None,
347 text: Some(value.clone()),
348 blob_base64: None,
349 },
350 Self::BlobBase64(value) => ValueWire {
351 null: false,
352 int: None,
353 float: None,
354 text: None,
355 blob_base64: Some(value.clone()),
356 },
357 };
358 wire.serialize(serializer)
359 }
360 }
361
362 impl<'de> Deserialize<'de> for Value {
363 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
364 where
365 D: Deserializer<'de>,
366 {
367 let wire = ValueWire::deserialize(deserializer)?;
368 let variants = usize::from(wire.null)
369 + usize::from(wire.int.is_some())
370 + usize::from(wire.float.is_some())
371 + usize::from(wire.text.is_some())
372 + usize::from(wire.blob_base64.is_some());
373 if variants != 1 {
374 return Err(serde::de::Error::custom(
375 "SQLite value must contain exactly one typed field",
376 ));
377 }
378 if wire.null {
379 return Ok(Self::Null);
380 }
381 if let Some(value) = wire.int {
382 return Ok(Self::Integer(value));
383 }
384 if let Some(value) = wire.float {
385 return Ok(Self::Float(value));
386 }
387 if let Some(value) = wire.text {
388 return Ok(Self::Text(value));
389 }
390 Ok(Self::BlobBase64(wire.blob_base64.unwrap_or_default()))
391 }
392 }
393
394 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
395 #[serde(deny_unknown_fields)]
396 pub struct ExecRequest {
397 pub store_id: String,
398 #[serde(default, skip_serializing_if = "String::is_empty")]
399 pub database: String,
400 pub sql: String,
401 #[serde(default, skip_serializing_if = "Vec::is_empty")]
402 pub args: Vec<Value>,
403 #[serde(default, skip_serializing_if = "Option::is_none")]
404 pub timeout_ms: Option<u64>,
405 }
406
407 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
408 #[serde(deny_unknown_fields)]
409 pub struct ExecResponse {
410 pub ok: bool,
411 pub database: String,
412 pub rows_affected: i64,
413 #[serde(default)]
414 pub last_insert_id: i64,
415 pub usage: Usage,
416 }
417
418 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
419 #[serde(deny_unknown_fields)]
420 pub struct QueryRequest {
421 pub store_id: String,
422 #[serde(default, skip_serializing_if = "String::is_empty")]
423 pub database: String,
424 pub sql: String,
425 #[serde(default, skip_serializing_if = "Vec::is_empty")]
426 pub args: Vec<Value>,
427 #[serde(default, skip_serializing_if = "Option::is_none")]
428 pub max_rows: Option<u32>,
429 #[serde(default, skip_serializing_if = "Option::is_none")]
430 pub max_response_bytes: Option<u64>,
431 #[serde(default, skip_serializing_if = "Option::is_none")]
432 pub timeout_ms: Option<u64>,
433 }
434
435 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
436 #[serde(deny_unknown_fields)]
437 pub struct QueryResponse {
438 pub ok: bool,
439 pub database: String,
440 pub columns: Vec<String>,
441 pub rows: Vec<Vec<Value>>,
442 pub usage: Usage,
443 }
444
445 pub fn exec(request: ExecRequest) -> Result<ExecResponse, WorkerError> {
446 call_broker(
447 "storage.sqlite",
448 &OperationRequest {
449 operation: "exec",
450 request: &request,
451 },
452 )
453 }
454
455 pub fn query(request: QueryRequest) -> Result<QueryResponse, WorkerError> {
456 call_broker(
457 "storage.sqlite",
458 &OperationRequest {
459 operation: "query",
460 request: &request,
461 },
462 )
463 }
464 }
465}
466
467#[cfg(test)]
468mod tests {
469 use super::*;
470 use serde_json::json;
471
472 #[test]
473 fn sqlite_values_are_exactly_typed() {
474 let values = vec![
475 storage::sqlite::Value::Null,
476 storage::sqlite::Value::Integer(7),
477 storage::sqlite::Value::Float(1.5),
478 storage::sqlite::Value::Text("memo".to_string()),
479 storage::sqlite::Value::BlobBase64("AAE=".to_string()),
480 ];
481 let encoded = serde_json::to_value(&values).unwrap();
482 let decoded: Vec<storage::sqlite::Value> = serde_json::from_value(encoded).unwrap();
483 assert_eq!(decoded, values);
484 assert!(
485 serde_json::from_value::<storage::sqlite::Value>(json!({"int": 1, "text": "x"}))
486 .is_err()
487 );
488 assert!(
489 serde_json::from_value::<storage::sqlite::Value>(
490 json!({"text": "x", "token": "secret"})
491 )
492 .is_err()
493 );
494 }
495
496 #[test]
497 fn storage_operation_is_an_argument_of_the_single_control_call() {
498 let encoded = serde_json::to_value(OperationRequest {
499 operation: "query",
500 request: &storage::sqlite::QueryRequest {
501 store_id: "memos".to_string(),
502 database: "memos.sqlite".to_string(),
503 sql: "SELECT 1".to_string(),
504 args: vec![],
505 max_rows: Some(1),
506 max_response_bytes: Some(4096),
507 timeout_ms: Some(1000),
508 },
509 })
510 .unwrap();
511 assert_eq!(encoded["operation"], "query");
512 assert_eq!(encoded["store_id"], "memos");
513 assert!(encoded.get("plugin_api").is_none());
514 }
515
516 #[test]
517 fn typed_success_responses_reject_unknown_fields() {
518 let response = json!({
519 "ok": true,
520 "database": "notes.sqlite",
521 "columns": ["title"],
522 "rows": [[{"text": "Launch"}]],
523 "usage": {
524 "plugin_instance_id": "plugini_1",
525 "store_id": "notes",
526 "usage_bytes": 10,
527 "quota_bytes": 100,
528 "usage_files": 1,
529 "quota_files": 4
530 },
531 "handle_grant_token": "secret"
532 });
533 assert!(serde_json::from_value::<storage::sqlite::QueryResponse>(response).is_err());
534 }
535
536 #[test]
537 fn operation_specific_storage_responses_reject_cross_operation_fields() {
538 let usage = json!({
539 "plugin_instance_id": "plugini_1",
540 "store_id": "workspace",
541 "usage_bytes": 10,
542 "quota_bytes": 100,
543 "usage_files": 1,
544 "quota_files": 4
545 });
546 let cases = [
547 serde_json::to_vec(&json!({
548 "ok": true,
549 "path": "notes/a.txt",
550 "data_base64": "YQ==",
551 "size_bytes": 1,
552 "entries": [],
553 "usage": usage.clone()
554 }))
555 .unwrap(),
556 serde_json::to_vec(&json!({
557 "ok": true,
558 "path": "notes/a.txt",
559 "size_bytes": 1,
560 "data_base64": "YQ==",
561 "usage": usage.clone()
562 }))
563 .unwrap(),
564 serde_json::to_vec(&json!({
565 "ok": true,
566 "path": "notes/a.txt",
567 "usage": usage.clone()
568 }))
569 .unwrap(),
570 serde_json::to_vec(&json!({
571 "ok": true,
572 "path": "notes",
573 "entries": [],
574 "data_base64": "YQ==",
575 "usage": usage.clone()
576 }))
577 .unwrap(),
578 ];
579 assert!(serde_json::from_slice::<storage::files::ReadResponse>(&cases[0]).is_err());
580 assert!(serde_json::from_slice::<storage::files::WriteResponse>(&cases[1]).is_err());
581 assert!(serde_json::from_slice::<storage::files::DeleteResponse>(&cases[2]).is_err());
582 assert!(serde_json::from_slice::<storage::files::ListResponse>(&cases[3]).is_err());
583
584 let kv_get = serde_json::to_vec(&json!({
585 "ok": true,
586 "key": "theme",
587 "value_base64": "ZGFyaw==",
588 "size_bytes": 4,
589 "entries": [],
590 "usage": usage.clone()
591 }))
592 .unwrap();
593 let kv_put = serde_json::to_vec(&json!({
594 "ok": true,
595 "key": "theme",
596 "size_bytes": 4,
597 "value_base64": "ZGFyaw==",
598 "usage": usage.clone()
599 }))
600 .unwrap();
601 let kv_delete = serde_json::to_vec(&json!({
602 "ok": true,
603 "key": "theme",
604 "usage": usage.clone()
605 }))
606 .unwrap();
607 let kv_list = serde_json::to_vec(&json!({
608 "ok": true,
609 "prefix": "settings/",
610 "entries": [],
611 "value_base64": "ZGFyaw==",
612 "usage": usage.clone()
613 }))
614 .unwrap();
615 assert!(serde_json::from_slice::<storage::kv::GetResponse>(&kv_get).is_err());
616 assert!(serde_json::from_slice::<storage::kv::PutResponse>(&kv_put).is_err());
617 assert!(serde_json::from_slice::<storage::kv::DeleteResponse>(&kv_delete).is_err());
618 assert!(serde_json::from_slice::<storage::kv::ListResponse>(&kv_list).is_err());
619
620 let sqlite_exec = serde_json::to_vec(&json!({
621 "ok": true,
622 "database": "notes.sqlite",
623 "rows_affected": 1,
624 "columns": [],
625 "rows": [],
626 "usage": usage.clone()
627 }))
628 .unwrap();
629 let sqlite_query = serde_json::to_vec(&json!({
630 "ok": true,
631 "database": "notes.sqlite",
632 "columns": [],
633 "rows": [],
634 "rows_affected": 1,
635 "usage": usage
636 }))
637 .unwrap();
638 assert!(serde_json::from_slice::<storage::sqlite::ExecResponse>(&sqlite_exec).is_err());
639 assert!(serde_json::from_slice::<storage::sqlite::QueryResponse>(&sqlite_query).is_err());
640 }
641
642 #[test]
643 fn broker_errors_preserve_the_plugin_api_code_and_message() {
644 let error = worker_error(Error {
645 code: crate::ErrorCode::PermissionDenied,
646 message: "blocked".to_string(),
647 retryable: false,
648 details: serde_json::Value::Null,
649 });
650 assert_eq!(error.code, "PERMISSION_DENIED");
651 assert_eq!(error.message, "blocked");
652 }
653}