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> {
458 call_broker(
459 "storage.sqlite",
460 &OperationRequest {
461 operation: "query",
462 request: &request,
463 },
464 )
465 }
466 }
467}
468
469#[cfg(test)]
470mod tests {
471 use super::*;
472 use serde_json::json;
473
474 #[test]
475 fn sqlite_values_are_exactly_typed() {
476 let values = vec![
477 storage::sqlite::Value::Null,
478 storage::sqlite::Value::Integer(7),
479 storage::sqlite::Value::Float(1.5),
480 storage::sqlite::Value::Text("memo".to_string()),
481 storage::sqlite::Value::BlobBase64("AAE=".to_string()),
482 ];
483 let encoded = serde_json::to_value(&values).unwrap();
484 let decoded: Vec<storage::sqlite::Value> = serde_json::from_value(encoded).unwrap();
485 assert_eq!(decoded, values);
486 assert!(
487 serde_json::from_value::<storage::sqlite::Value>(json!({"int": 1, "text": "x"}))
488 .is_err()
489 );
490 assert!(
491 serde_json::from_value::<storage::sqlite::Value>(
492 json!({"text": "x", "token": "secret"})
493 )
494 .is_err()
495 );
496 }
497
498 #[test]
499 fn storage_operation_is_an_argument_of_the_single_control_call() {
500 let encoded = serde_json::to_value(OperationRequest {
501 operation: "query",
502 request: &storage::sqlite::QueryRequest {
503 store_id: "memos".to_string(),
504 database: "memos.sqlite".to_string(),
505 sql: "SELECT 1".to_string(),
506 args: vec![],
507 max_rows: Some(1),
508 max_response_bytes: Some(4096),
509 timeout_ms: Some(1000),
510 },
511 })
512 .unwrap();
513 assert_eq!(encoded["operation"], "query");
514 assert_eq!(encoded["store_id"], "memos");
515 assert!(encoded.get("plugin_api").is_none());
516 }
517
518 #[test]
519 fn typed_success_responses_reject_unknown_fields() {
520 let response = json!({
521 "ok": true,
522 "database": "notes.sqlite",
523 "columns": ["title"],
524 "rows": [[{"text": "Launch"}]],
525 "usage": {
526 "plugin_instance_id": "plugini_1",
527 "store_id": "notes",
528 "usage_bytes": 10,
529 "quota_bytes": 100,
530 "usage_files": 1,
531 "quota_files": 4
532 },
533 "handle_grant_token": "secret"
534 });
535 assert!(serde_json::from_value::<storage::sqlite::QueryResponse>(response).is_err());
536 }
537
538 #[test]
539 fn operation_specific_storage_responses_reject_cross_operation_fields() {
540 let usage = json!({
541 "plugin_instance_id": "plugini_1",
542 "store_id": "workspace",
543 "usage_bytes": 10,
544 "quota_bytes": 100,
545 "usage_files": 1,
546 "quota_files": 4
547 });
548 let cases = [
549 serde_json::to_vec(&json!({
550 "ok": true,
551 "path": "notes/a.txt",
552 "data_base64": "YQ==",
553 "size_bytes": 1,
554 "entries": [],
555 "usage": usage.clone()
556 }))
557 .unwrap(),
558 serde_json::to_vec(&json!({
559 "ok": true,
560 "path": "notes/a.txt",
561 "size_bytes": 1,
562 "data_base64": "YQ==",
563 "usage": usage.clone()
564 }))
565 .unwrap(),
566 serde_json::to_vec(&json!({
567 "ok": true,
568 "path": "notes/a.txt",
569 "usage": usage.clone()
570 }))
571 .unwrap(),
572 serde_json::to_vec(&json!({
573 "ok": true,
574 "path": "notes",
575 "entries": [],
576 "data_base64": "YQ==",
577 "usage": usage.clone()
578 }))
579 .unwrap(),
580 ];
581 assert!(serde_json::from_slice::<storage::files::ReadResponse>(&cases[0]).is_err());
582 assert!(serde_json::from_slice::<storage::files::WriteResponse>(&cases[1]).is_err());
583 assert!(serde_json::from_slice::<storage::files::DeleteResponse>(&cases[2]).is_err());
584 assert!(serde_json::from_slice::<storage::files::ListResponse>(&cases[3]).is_err());
585
586 let kv_get = serde_json::to_vec(&json!({
587 "ok": true,
588 "key": "theme",
589 "value_base64": "ZGFyaw==",
590 "size_bytes": 4,
591 "entries": [],
592 "usage": usage.clone()
593 }))
594 .unwrap();
595 let kv_put = serde_json::to_vec(&json!({
596 "ok": true,
597 "key": "theme",
598 "size_bytes": 4,
599 "value_base64": "ZGFyaw==",
600 "usage": usage.clone()
601 }))
602 .unwrap();
603 let kv_delete = serde_json::to_vec(&json!({
604 "ok": true,
605 "key": "theme",
606 "usage": usage.clone()
607 }))
608 .unwrap();
609 let kv_list = serde_json::to_vec(&json!({
610 "ok": true,
611 "prefix": "settings/",
612 "entries": [],
613 "value_base64": "ZGFyaw==",
614 "usage": usage.clone()
615 }))
616 .unwrap();
617 assert!(serde_json::from_slice::<storage::kv::GetResponse>(&kv_get).is_err());
618 assert!(serde_json::from_slice::<storage::kv::PutResponse>(&kv_put).is_err());
619 assert!(serde_json::from_slice::<storage::kv::DeleteResponse>(&kv_delete).is_err());
620 assert!(serde_json::from_slice::<storage::kv::ListResponse>(&kv_list).is_err());
621
622 let sqlite_exec = serde_json::to_vec(&json!({
623 "ok": true,
624 "database": "notes.sqlite",
625 "rows_affected": 1,
626 "columns": [],
627 "rows": [],
628 "usage": usage.clone()
629 }))
630 .unwrap();
631 let sqlite_query = serde_json::to_vec(&json!({
632 "ok": true,
633 "database": "notes.sqlite",
634 "columns": [],
635 "rows": [],
636 "rows_affected": 1,
637 "usage": usage
638 }))
639 .unwrap();
640 assert!(serde_json::from_slice::<storage::sqlite::ExecResponse>(&sqlite_exec).is_err());
641 assert!(serde_json::from_slice::<storage::sqlite::QueryResponse>(&sqlite_query).is_err());
642 }
643
644 #[test]
645 fn broker_errors_preserve_the_plugin_api_code_and_message() {
646 let error = worker_error(Error {
647 code: crate::ErrorCode::PermissionDenied,
648 message: "blocked".to_string(),
649 retryable: false,
650 details: serde_json::Value::Null,
651 });
652 assert_eq!(error.code, "PERMISSION_DENIED");
653 assert_eq!(error.message, "blocked");
654 }
655}