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 conditional {
293 use super::*;
294 #[derive(Debug, Clone, Serialize, Deserialize)]
295 #[serde(deny_unknown_fields)]
296 pub struct KVSnapshot {
297 pub ok: bool,
298 pub key: String,
299 pub value_base64: String,
300 pub size_bytes: i64,
301 pub usage: Usage,
302 pub etag: String,
303 }
304 #[derive(Debug, Clone, Serialize, Deserialize)]
305 #[serde(deny_unknown_fields)]
306 pub struct KVPutResult {
307 pub ok: bool,
308 pub key: String,
309 pub size_bytes: i64,
310 pub usage: Usage,
311 pub etag: String,
312 }
313 pub fn get(store_id: &str, key: &str, max_bytes: u64) -> Result<KVSnapshot, WorkerError> {
314 call_broker(
315 "storage.kv",
316 &serde_json::json!({"operation":"get","store_id":store_id,"key":key,"max_bytes":max_bytes,"include_etag":true}),
317 )
318 }
319 pub fn put(
320 store_id: &str,
321 key: &str,
322 value_base64: &str,
323 expected_etag: &str,
324 ) -> Result<KVPutResult, WorkerError> {
325 call_broker(
326 "storage.kv",
327 &serde_json::json!({"operation":"put","store_id":store_id,"key":key,"value_base64":value_base64,"expected_etag":expected_etag,"include_etag":true}),
328 )
329 }
330 pub fn delete(
331 store_id: &str,
332 key: &str,
333 expected_etag: &str,
334 ) -> Result<kv::DeleteResponse, WorkerError> {
335 call_broker(
336 "storage.kv",
337 &serde_json::json!({"operation":"delete","store_id":store_id,"key":key,"expected_etag":expected_etag}),
338 )
339 }
340 #[derive(Debug, Clone, Serialize, Deserialize)]
341 #[serde(deny_unknown_fields)]
342 pub struct KVPage {
343 pub ok: bool,
344 pub prefix: String,
345 pub entries: Vec<kv::Entry>,
346 pub usage: Usage,
347 pub next_cursor: String,
348 }
349 pub fn list(
350 store_id: &str,
351 prefix: &str,
352 cursor: &str,
353 max_entries: u32,
354 ) -> Result<KVPage, WorkerError> {
355 call_broker(
356 "storage.kv",
357 &serde_json::json!({"operation":"list","store_id":store_id,"prefix":prefix,"cursor":cursor,"max_entries":max_entries,"include_cursor":true}),
358 )
359 }
360 #[derive(Debug, Clone, Serialize, Deserialize)]
361 #[serde(deny_unknown_fields)]
362 pub struct FilePage {
363 pub ok: bool,
364 pub path: String,
365 pub entries: Vec<files::Entry>,
366 pub usage: Usage,
367 pub next_cursor: String,
368 }
369 pub fn list_files(
370 store_id: &str,
371 path: &str,
372 cursor: &str,
373 max_entries: u32,
374 ) -> Result<FilePage, WorkerError> {
375 call_broker(
376 "storage.files",
377 &serde_json::json!({"operation":"list","store_id":store_id,"path":path,"cursor":cursor,"max_entries":max_entries,"include_cursor":true}),
378 )
379 }
380 }
381
382 pub mod sqlite {
383 use super::*;
384
385 #[derive(Debug, Clone, PartialEq)]
386 pub enum Value {
387 Null,
388 Integer(i64),
389 Float(f64),
390 Text(String),
391 BlobBase64(String),
392 }
393
394 #[derive(Serialize, Deserialize)]
395 #[serde(deny_unknown_fields)]
396 struct ValueWire {
397 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
398 null: bool,
399 #[serde(default, skip_serializing_if = "Option::is_none")]
400 int: Option<i64>,
401 #[serde(default, skip_serializing_if = "Option::is_none")]
402 float: Option<f64>,
403 #[serde(default, skip_serializing_if = "Option::is_none")]
404 text: Option<String>,
405 #[serde(default, skip_serializing_if = "Option::is_none")]
406 blob_base64: Option<String>,
407 }
408
409 impl Serialize for Value {
410 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
411 where
412 S: Serializer,
413 {
414 let wire = match self {
415 Self::Null => ValueWire {
416 null: true,
417 int: None,
418 float: None,
419 text: None,
420 blob_base64: None,
421 },
422 Self::Integer(value) => ValueWire {
423 null: false,
424 int: Some(*value),
425 float: None,
426 text: None,
427 blob_base64: None,
428 },
429 Self::Float(value) => ValueWire {
430 null: false,
431 int: None,
432 float: Some(*value),
433 text: None,
434 blob_base64: None,
435 },
436 Self::Text(value) => ValueWire {
437 null: false,
438 int: None,
439 float: None,
440 text: Some(value.clone()),
441 blob_base64: None,
442 },
443 Self::BlobBase64(value) => ValueWire {
444 null: false,
445 int: None,
446 float: None,
447 text: None,
448 blob_base64: Some(value.clone()),
449 },
450 };
451 wire.serialize(serializer)
452 }
453 }
454
455 impl<'de> Deserialize<'de> for Value {
456 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
457 where
458 D: Deserializer<'de>,
459 {
460 let wire = ValueWire::deserialize(deserializer)?;
461 let variants = usize::from(wire.null)
462 + usize::from(wire.int.is_some())
463 + usize::from(wire.float.is_some())
464 + usize::from(wire.text.is_some())
465 + usize::from(wire.blob_base64.is_some());
466 if variants != 1 {
467 return Err(serde::de::Error::custom(
468 "SQLite value must contain exactly one typed field",
469 ));
470 }
471 if wire.null {
472 return Ok(Self::Null);
473 }
474 if let Some(value) = wire.int {
475 return Ok(Self::Integer(value));
476 }
477 if let Some(value) = wire.float {
478 return Ok(Self::Float(value));
479 }
480 if let Some(value) = wire.text {
481 return Ok(Self::Text(value));
482 }
483 Ok(Self::BlobBase64(wire.blob_base64.unwrap_or_default()))
484 }
485 }
486
487 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
488 #[serde(deny_unknown_fields)]
489 pub struct Statement {
490 pub sql: String,
491 #[serde(default, skip_serializing_if = "Vec::is_empty")]
492 pub args: Vec<Value>,
493 }
494 pub fn transaction(
497 store_id: &str,
498 database: &str,
499 statements: &[Statement],
500 timeout_ms: u64,
501 ) -> Result<ExecResponse, WorkerError> {
502 if statements.is_empty() || statements.len() > 64 {
503 return Err(WorkerError::invalid_request(
504 "transaction requires 1 to 64 statements",
505 ));
506 }
507 call_broker(
508 "storage.sqlite",
509 &serde_json::json!({"operation":"exec","store_id":store_id,"database":database,"statements":statements,"timeout_ms":timeout_ms}),
510 )
511 }
512
513 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
514 #[serde(deny_unknown_fields)]
515 pub struct ExecRequest {
516 pub store_id: String,
517 #[serde(default, skip_serializing_if = "String::is_empty")]
518 pub database: String,
519 pub sql: String,
520 #[serde(default, skip_serializing_if = "Vec::is_empty")]
521 pub args: Vec<Value>,
522 #[serde(default, skip_serializing_if = "Option::is_none")]
523 pub timeout_ms: Option<u64>,
524 }
525
526 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
527 #[serde(deny_unknown_fields)]
528 pub struct ExecResponse {
529 pub ok: bool,
530 pub database: String,
531 pub rows_affected: i64,
532 #[serde(default)]
533 pub last_insert_id: i64,
534 pub usage: Usage,
535 }
536
537 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
538 #[serde(deny_unknown_fields)]
539 pub struct QueryRequest {
540 pub store_id: String,
541 #[serde(default, skip_serializing_if = "String::is_empty")]
542 pub database: String,
543 pub sql: String,
544 #[serde(default, skip_serializing_if = "Vec::is_empty")]
545 pub args: Vec<Value>,
546 #[serde(default, skip_serializing_if = "Option::is_none")]
547 pub max_rows: Option<u32>,
548 #[serde(default, skip_serializing_if = "Option::is_none")]
549 pub max_response_bytes: Option<u64>,
550 #[serde(default, skip_serializing_if = "Option::is_none")]
551 pub timeout_ms: Option<u64>,
552 }
553
554 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
555 #[serde(deny_unknown_fields)]
556 pub struct QueryResponse {
557 pub ok: bool,
558 pub database: String,
559 pub columns: Vec<String>,
560 pub rows: Vec<Vec<Value>>,
561 pub usage: Usage,
562 }
563
564 pub fn exec(request: ExecRequest) -> Result<ExecResponse, WorkerError> {
565 call_broker(
566 "storage.sqlite",
567 &OperationRequest {
568 operation: "exec",
569 request: &request,
570 },
571 )
572 }
573
574 pub fn query(request: QueryRequest) -> Result<QueryResponse, WorkerError> {
577 call_broker(
578 "storage.sqlite",
579 &OperationRequest {
580 operation: "query",
581 request: &request,
582 },
583 )
584 }
585 }
586}
587
588#[cfg(test)]
589mod tests {
590 use super::*;
591 use serde_json::json;
592
593 #[test]
594 fn sqlite_values_are_exactly_typed() {
595 let values = vec![
596 storage::sqlite::Value::Null,
597 storage::sqlite::Value::Integer(7),
598 storage::sqlite::Value::Float(1.5),
599 storage::sqlite::Value::Text("memo".to_string()),
600 storage::sqlite::Value::BlobBase64("AAE=".to_string()),
601 ];
602 let encoded = serde_json::to_value(&values).unwrap();
603 let decoded: Vec<storage::sqlite::Value> = serde_json::from_value(encoded).unwrap();
604 assert_eq!(decoded, values);
605 assert!(
606 serde_json::from_value::<storage::sqlite::Value>(json!({"int": 1, "text": "x"}))
607 .is_err()
608 );
609 assert!(
610 serde_json::from_value::<storage::sqlite::Value>(
611 json!({"text": "x", "token": "secret"})
612 )
613 .is_err()
614 );
615 }
616
617 #[test]
618 fn storage_operation_is_an_argument_of_the_single_control_call() {
619 let encoded = serde_json::to_value(OperationRequest {
620 operation: "query",
621 request: &storage::sqlite::QueryRequest {
622 store_id: "memos".to_string(),
623 database: "memos.sqlite".to_string(),
624 sql: "SELECT 1".to_string(),
625 args: vec![],
626 max_rows: Some(1),
627 max_response_bytes: Some(4096),
628 timeout_ms: Some(1000),
629 },
630 })
631 .unwrap();
632 assert_eq!(encoded["operation"], "query");
633 assert_eq!(encoded["store_id"], "memos");
634 assert!(encoded.get("plugin_api").is_none());
635 }
636
637 #[test]
638 fn typed_success_responses_reject_unknown_fields() {
639 let response = json!({
640 "ok": true,
641 "database": "notes.sqlite",
642 "columns": ["title"],
643 "rows": [[{"text": "Launch"}]],
644 "usage": {
645 "plugin_instance_id": "plugini_1",
646 "store_id": "notes",
647 "usage_bytes": 10,
648 "quota_bytes": 100,
649 "usage_files": 1,
650 "quota_files": 4
651 },
652 "handle_grant_token": "secret"
653 });
654 assert!(serde_json::from_value::<storage::sqlite::QueryResponse>(response).is_err());
655 }
656
657 #[test]
658 fn operation_specific_storage_responses_reject_cross_operation_fields() {
659 let usage = json!({
660 "plugin_instance_id": "plugini_1",
661 "store_id": "workspace",
662 "usage_bytes": 10,
663 "quota_bytes": 100,
664 "usage_files": 1,
665 "quota_files": 4
666 });
667 let cases = [
668 serde_json::to_vec(&json!({
669 "ok": true,
670 "path": "notes/a.txt",
671 "data_base64": "YQ==",
672 "size_bytes": 1,
673 "entries": [],
674 "usage": usage.clone()
675 }))
676 .unwrap(),
677 serde_json::to_vec(&json!({
678 "ok": true,
679 "path": "notes/a.txt",
680 "size_bytes": 1,
681 "data_base64": "YQ==",
682 "usage": usage.clone()
683 }))
684 .unwrap(),
685 serde_json::to_vec(&json!({
686 "ok": true,
687 "path": "notes/a.txt",
688 "usage": usage.clone()
689 }))
690 .unwrap(),
691 serde_json::to_vec(&json!({
692 "ok": true,
693 "path": "notes",
694 "entries": [],
695 "data_base64": "YQ==",
696 "usage": usage.clone()
697 }))
698 .unwrap(),
699 ];
700 assert!(serde_json::from_slice::<storage::files::ReadResponse>(&cases[0]).is_err());
701 assert!(serde_json::from_slice::<storage::files::WriteResponse>(&cases[1]).is_err());
702 assert!(serde_json::from_slice::<storage::files::DeleteResponse>(&cases[2]).is_err());
703 assert!(serde_json::from_slice::<storage::files::ListResponse>(&cases[3]).is_err());
704
705 let kv_get = serde_json::to_vec(&json!({
706 "ok": true,
707 "key": "theme",
708 "value_base64": "ZGFyaw==",
709 "size_bytes": 4,
710 "entries": [],
711 "usage": usage.clone()
712 }))
713 .unwrap();
714 let kv_put = serde_json::to_vec(&json!({
715 "ok": true,
716 "key": "theme",
717 "size_bytes": 4,
718 "value_base64": "ZGFyaw==",
719 "usage": usage.clone()
720 }))
721 .unwrap();
722 let kv_delete = serde_json::to_vec(&json!({
723 "ok": true,
724 "key": "theme",
725 "usage": usage.clone()
726 }))
727 .unwrap();
728 let kv_list = serde_json::to_vec(&json!({
729 "ok": true,
730 "prefix": "settings/",
731 "entries": [],
732 "value_base64": "ZGFyaw==",
733 "usage": usage.clone()
734 }))
735 .unwrap();
736 assert!(serde_json::from_slice::<storage::kv::GetResponse>(&kv_get).is_err());
737 assert!(serde_json::from_slice::<storage::kv::PutResponse>(&kv_put).is_err());
738 assert!(serde_json::from_slice::<storage::kv::DeleteResponse>(&kv_delete).is_err());
739 assert!(serde_json::from_slice::<storage::kv::ListResponse>(&kv_list).is_err());
740
741 let sqlite_exec = serde_json::to_vec(&json!({
742 "ok": true,
743 "database": "notes.sqlite",
744 "rows_affected": 1,
745 "columns": [],
746 "rows": [],
747 "usage": usage.clone()
748 }))
749 .unwrap();
750 let sqlite_query = serde_json::to_vec(&json!({
751 "ok": true,
752 "database": "notes.sqlite",
753 "columns": [],
754 "rows": [],
755 "rows_affected": 1,
756 "usage": usage
757 }))
758 .unwrap();
759 assert!(serde_json::from_slice::<storage::sqlite::ExecResponse>(&sqlite_exec).is_err());
760 assert!(serde_json::from_slice::<storage::sqlite::QueryResponse>(&sqlite_query).is_err());
761 }
762
763 #[test]
764 fn broker_errors_preserve_the_plugin_api_code_and_message() {
765 let error = worker_error(Error {
766 code: crate::ErrorCode::PermissionDenied,
767 message: "blocked".to_string(),
768 retryable: false,
769 details: serde_json::Value::Null,
770 });
771 assert_eq!(error.code, "PERMISSION_DENIED");
772 assert_eq!(error.message, "blocked");
773 }
774}