1pub mod v1 {
4 tonic::include_proto!("keldra.v1");
5}
6
7#[derive(Clone, Copy, Debug, Eq, PartialEq)]
9pub enum PredicateExpressionError {
10 EmptyConjunction,
11 EmptyDisjunction,
12}
13
14impl std::fmt::Display for PredicateExpressionError {
15 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
16 match self {
17 Self::EmptyConjunction => {
18 formatter.write_str("a predicate conjunction requires at least one child")
19 }
20 Self::EmptyDisjunction => {
21 formatter.write_str("a predicate disjunction requires at least one child")
22 }
23 }
24 }
25}
26
27impl std::error::Error for PredicateExpressionError {}
28
29impl v1::IndexPredicateExpression {
30 pub fn leaf(predicate: v1::IndexPredicate) -> Self {
32 Self {
33 expression: Some(v1::index_predicate_expression::Expression::Predicate(
34 predicate,
35 )),
36 }
37 }
38
39 pub fn all(
41 expressions: impl IntoIterator<Item = Self>,
42 ) -> Result<Self, PredicateExpressionError> {
43 let expressions = expressions.into_iter().collect::<Vec<_>>();
44 if expressions.is_empty() {
45 return Err(PredicateExpressionError::EmptyConjunction);
46 }
47 Ok(Self {
48 expression: Some(v1::index_predicate_expression::Expression::Conjunction(
49 v1::IndexPredicateConjunction { expressions },
50 )),
51 })
52 }
53
54 pub fn any(
56 expressions: impl IntoIterator<Item = Self>,
57 ) -> Result<Self, PredicateExpressionError> {
58 let expressions = expressions.into_iter().collect::<Vec<_>>();
59 if expressions.is_empty() {
60 return Err(PredicateExpressionError::EmptyDisjunction);
61 }
62 Ok(Self {
63 expression: Some(v1::index_predicate_expression::Expression::Disjunction(
64 v1::IndexPredicateDisjunction { expressions },
65 )),
66 })
67 }
68
69 pub fn negated(self) -> Self {
71 Self {
72 expression: Some(v1::index_predicate_expression::Expression::Negation(
73 Box::new(v1::IndexPredicateNegation {
74 expression: Some(Box::new(self)),
75 }),
76 )),
77 }
78 }
79}
80
81#[cfg(test)]
82mod tests {
83 use prost::Message;
84
85 use super::v1::{
86 CloneObjectRequest, DeletedObject, Durability, LinkObjectRequest, NeverExisted,
87 ObjectAddress, ObjectHead, PresentObject, PutIfVersionOperation, UnlinkObjectRequest,
88 clone_object_request, object_head,
89 };
90
91 #[test]
92 fn exact_path_states_are_distinct() {
93 let states = [
94 ObjectHead {
95 state: Some(object_head::State::Present(PresentObject {
96 version: 7,
97 content_hash: vec![7; 32],
98 content_length: 99,
99 content_type: String::new(),
100 })),
101 },
102 ObjectHead {
103 state: Some(object_head::State::Deleted(DeletedObject { version: 8 })),
104 },
105 ObjectHead {
106 state: Some(object_head::State::NeverExisted(NeverExisted {})),
107 },
108 ];
109
110 assert!(matches!(
111 &states[0].state,
112 Some(object_head::State::Present(_))
113 ));
114 assert!(matches!(
115 &states[1].state,
116 Some(object_head::State::Deleted(_))
117 ));
118 assert!(matches!(
119 &states[2].state,
120 Some(object_head::State::NeverExisted(_))
121 ));
122 }
123
124 #[test]
125 fn clone_object_wire_round_trip_preserves_both_identities_and_exact_cas() {
126 let request = CloneObjectRequest {
127 source: Some(ObjectAddress {
128 tenant: "tenant".into(),
129 bucket: "bucket".into(),
130 path: "source".into(),
131 }),
132 source_version: 17,
133 destination: Some(ObjectAddress {
134 tenant: "tenant".into(),
135 bucket: "bucket".into(),
136 path: "destination".into(),
137 }),
138 command_id: "clone-17".into(),
139 durability: Durability::Replicated as i32,
140 operation: Some(clone_object_request::Operation::PutIfVersion(
141 PutIfVersionOperation {
142 expected_version: 11,
143 },
144 )),
145 };
146
147 let decoded = CloneObjectRequest::decode(request.encode_to_vec().as_slice()).unwrap();
148 assert_eq!(decoded, request);
149 }
150
151 #[test]
152 fn object_link_wire_contract_keeps_link_and_target_distinct() {
153 let link = ObjectAddress {
154 tenant: "tenant".into(),
155 bucket: "bucket".into(),
156 path: "alias".into(),
157 };
158 let request = LinkObjectRequest {
159 link: Some(link.clone()),
160 target: Some(ObjectAddress {
161 tenant: "tenant".into(),
162 bucket: "bucket".into(),
163 path: "target".into(),
164 }),
165 command_id: "link-1".into(),
166 durability: Durability::Replicated as i32,
167 };
168 assert_eq!(
169 LinkObjectRequest::decode(request.encode_to_vec().as_slice()).unwrap(),
170 request
171 );
172 let unlink = UnlinkObjectRequest {
173 link: Some(link),
174 command_id: "unlink-1".into(),
175 durability: Durability::Local as i32,
176 };
177 assert_eq!(
178 UnlinkObjectRequest::decode(unlink.encode_to_vec().as_slice()).unwrap(),
179 unlink
180 );
181 }
182
183 #[test]
184 fn schema_keeps_removed_capabilities_out() {
185 let schema = include_str!("../proto/keldra.proto").to_ascii_lowercase();
186 for forbidden in [
187 "rpc uploadblob",
188 "rpc publishobject",
189 "rpc putobject",
190 "message blobref",
191 "rpc listprefix",
192 "rpc begintransaction",
193 "rpc committransaction",
194 "personaldb",
195 ] {
196 assert!(!schema.contains(forbidden), "schema contains `{forbidden}`");
197 }
198
199 assert!(schema.contains("executor_nomination_log_index"));
200 assert!(schema.contains("commit_log_index"));
201 assert!(schema.contains("immutable_path_prefixes"));
202 assert!(schema.contains("program_only_path_prefixes"));
203 assert!(!schema.contains("rpc registerprogram"));
204 assert!(!schema.contains("message registerprogram"));
205 assert!(schema.contains("objectaddress program"));
206 assert!(schema.contains("_keldra/programs/{name}@{version}"));
207 assert!(schema.contains("rpc startput(putheader) returns (puttoken)"));
208 assert!(schema.contains("rpc put(stream putrequest) returns (puttoken)"));
209 assert!(schema.contains("rpc putend(puttoken) returns (mutationreceipt)"));
210 for rpc in [
211 "rpc createindex(createindexrequest)",
212 "rpc updateindex(updateindexrequest)",
213 "rpc getindex(getindexrequest)",
214 "rpc listindices(listindicesrequest)",
215 "rpc deleteindex(deleteindexrequest)",
216 "rpc queryindex(queryindexrequest)",
217 ] {
218 assert!(schema.contains(rpc), "schema is missing `{rpc}`");
219 }
220 assert!(schema.contains("index_kind_tensor"));
221 assert!(schema.contains("tensorindexspec tensor"));
222 assert!(schema.contains("tensorindexquery tensor"));
223
224 for rpc in [
225 "rpc exchangeclientcredentials",
226 "rpc provisiontenant",
227 "rpc createapplication",
228 "rpc rotateapplicationcredential",
229 "rpc disableapplicationcredential",
230 "rpc createbucket",
231 "rpc grantapplicationrole",
232 "rpc revokeapplicationrole",
233 "rpc putschema",
234 "rpc bindschema",
235 "rpc getbinding",
236 "rpc getschema",
237 "rpc mutatetuples",
238 "rpc readtuples",
239 "rpc checkpermission",
240 "rpc checkpermissions",
241 "rpc watchprefix",
242 "rpc listobjects",
243 "rpc cloneobject",
244 "rpc deleteversion",
245 "rpc listobjectversions",
246 "rpc setbucketversioning",
247 ] {
248 assert!(schema.contains(rpc), "schema is missing `{rpc}`");
249 }
250 for forbidden in [
251 "rpc createrealm",
252 "rpc deleterealm",
253 "rpc applyschema",
254 "zookie",
255 "caveat",
256 "publication_metadata",
257 "insecure_no_auth",
258 "api_token",
259 ] {
260 assert!(!schema.contains(forbidden), "schema contains `{forbidden}`");
261 }
262 }
263
264 #[test]
265 fn generated_index_client_is_publicly_exposed() {
266 let _: Option<
267 super::v1::index_service_client::IndexServiceClient<tonic::transport::Channel>,
268 > = None;
269 }
270
271 #[test]
272 fn typed_json_fields_have_explicit_type_cardinality_and_capabilities() {
273 use super::v1::{
274 IndexField, IndexFieldCapability, IndexFieldCardinality, KeywordIndexField, index_field,
275 };
276
277 let field = IndexField {
278 name: "document_id".into(),
279 json_pointer: "/id".into(),
280 cardinality: IndexFieldCardinality::Single as i32,
281 capabilities: vec![IndexFieldCapability::Exact as i32],
282 field_type: Some(index_field::FieldType::Keyword(KeywordIndexField {})),
283 };
284
285 assert!(matches!(
286 field.field_type,
287 Some(index_field::FieldType::Keyword(_))
288 ));
289 assert_eq!(field.capabilities, [IndexFieldCapability::Exact as i32]);
290
291 let date = super::v1::DateIndexField {
292 strftime_pattern: "%Y-%m-%d".into(),
293 };
294 assert!(matches!(
295 index_field::FieldType::Date(date),
296 index_field::FieldType::Date(value) if value.strftime_pattern == "%Y-%m-%d"
297 ));
298
299 let schema = include_str!("../proto/keldra.proto").to_ascii_lowercase();
300 assert!(schema.contains("rpc listindices(listindicesrequest)"));
301 assert!(schema.contains("repeated indexdefinition indices = 1"));
302 assert!(!schema.contains("listindexes"));
303 assert!(!schema.contains("fields_json"));
304 assert!(!schema.contains("bool multi_valued"));
305 }
306
307 #[test]
308 fn typed_json_queries_expose_fielded_text_facets_and_aggregates() {
309 use super::v1::index_predicate_expression::Expression;
310 use super::v1::{
311 IndexAggregateOperation, IndexAggregateRequest, IndexAggregateResult, IndexFacetBucket,
312 IndexFacetRequest, IndexFacetResult, IndexPredicate, IndexPredicateConjunction,
313 IndexPredicateExpression, IndexPredicateOperator, IndexQueryHit, QueryIndexResponse,
314 TypedJsonIndexQuery,
315 };
316
317 let query = TypedJsonIndexQuery {
318 predicate: Some(IndexPredicateExpression {
319 expression: Some(Expression::Conjunction(IndexPredicateConjunction {
320 expressions: [
321 IndexPredicateOperator::FullText,
322 IndexPredicateOperator::Phrase,
323 ]
324 .into_iter()
325 .map(|operator| IndexPredicateExpression {
326 expression: Some(Expression::Predicate(IndexPredicate {
327 field: "summary".into(),
328 operator: operator as i32,
329 values_json: vec![br#""memory safety""#.to_vec()],
330 })),
331 })
332 .collect(),
333 })),
334 }),
335 order: Vec::new(),
336 facets: vec![IndexFacetRequest {
337 field: "ecosystem".into(),
338 limit: 10,
339 }],
340 aggregates: vec![IndexAggregateRequest {
341 field: "severity".into(),
342 operation: IndexAggregateOperation::Average as i32,
343 }],
344 };
345 let response = QueryIndexResponse {
346 hits: vec![IndexQueryHit {
347 address: None,
348 object_version: 7,
349 score: Some(0.75),
350 }],
351 next_page_token: Vec::new(),
352 freshness: None,
353 facet_results: vec![IndexFacetResult {
354 field: "ecosystem".into(),
355 buckets: vec![IndexFacetBucket {
356 value_json: br#""cargo""#.to_vec(),
357 count: 4,
358 }],
359 }],
360 aggregate_results: vec![IndexAggregateResult {
361 field: "severity".into(),
362 operation: IndexAggregateOperation::Average as i32,
363 value_json: Some(b"7.5".to_vec()),
364 contributing_count: 4,
365 }],
366 };
367
368 assert_eq!(query.facets[0].limit, 10);
369 assert!(matches!(
370 query.predicate.unwrap().expression,
371 Some(Expression::Conjunction(_))
372 ));
373 assert_eq!(response.hits[0].object_version, 7);
374 assert_eq!(response.aggregate_results[0].contributing_count, 4);
375
376 let schema = include_str!("../proto/keldra.proto").to_ascii_lowercase();
377 assert!(!schema.contains("repeated indexpredicate predicates"));
378 assert!(schema.contains("indexpredicateexpression predicate = 5"));
379 assert!(schema.contains("indexpredicateexpression predicate = 2"));
380 }
381
382 #[test]
383 fn predicate_expression_helpers_reject_empty_boolean_operators() {
384 use super::PredicateExpressionError;
385 use super::v1::index_predicate_expression::Expression;
386 use super::v1::{IndexPredicate, IndexPredicateExpression, IndexPredicateOperator};
387
388 let leaf = IndexPredicateExpression::leaf(IndexPredicate {
389 field: "status".into(),
390 operator: IndexPredicateOperator::Exists as i32,
391 values_json: Vec::new(),
392 });
393 assert!(matches!(
394 IndexPredicateExpression::all([leaf.clone()])
395 .unwrap()
396 .expression,
397 Some(Expression::Conjunction(_))
398 ));
399 assert!(matches!(
400 IndexPredicateExpression::any([leaf.clone()])
401 .unwrap()
402 .expression,
403 Some(Expression::Disjunction(_))
404 ));
405 assert!(matches!(
406 leaf.negated().expression,
407 Some(Expression::Negation(_))
408 ));
409 assert_eq!(
410 IndexPredicateExpression::all(Vec::new()).unwrap_err(),
411 PredicateExpressionError::EmptyConjunction
412 );
413 assert_eq!(
414 IndexPredicateExpression::any(Vec::new()).unwrap_err(),
415 PredicateExpressionError::EmptyDisjunction
416 );
417 }
418
419 #[test]
420 fn generated_personaldb_client_is_publicly_exposed() {
421 let _: Option<
422 super::v1::personal_db_service_client::PersonalDbServiceClient<
423 tonic::transport::Channel,
424 >,
425 > = None;
426
427 let schema = include_str!("../proto/personaldb.proto").to_ascii_lowercase();
428 for rpc in [
429 "rpc creategroup(",
430 "rpc describegroup(",
431 "rpc listgroups(",
432 "rpc grantgrouprole(",
433 "rpc revokegrouprole(",
434 "rpc appendentry(",
435 "rpc materializeprojection(",
436 "rpc catchup(",
437 "rpc registersnapshot(",
438 "rpc getsnapshot(",
439 ] {
440 assert!(schema.contains(rpc), "PersonalDB schema is missing `{rpc}`");
441 }
442 }
443
444 #[test]
445 fn object_surface_has_only_explicit_typed_mutations() {
446 use super::v1::{
447 BulkOperation, BulkPutIfVersionRequest, CreateBucketRequest, DeleteIfVersionRequest,
448 DeleteRequest, DeleteVersionRequest, DeleteVersionResponse, Durability,
449 ListObjectsRequest, ListObjectsResponse, ObjectAddress, ObjectVersioning, PutHeader,
450 PutIfVersionOperation, PutRequest, PutToken, bulk_operation, put_header,
451 };
452
453 let address = Some(ObjectAddress {
454 tenant: "acme".into(),
455 bucket: "objects".into(),
456 path: "one".into(),
457 });
458 let header = PutHeader {
459 address: address.clone(),
460 content_type: "application/json".into(),
461 command_id: "command-1".into(),
462 durability: Durability::Local as i32,
463 operation: Some(put_header::Operation::PutIfVersion(PutIfVersionOperation {
464 expected_version: 8,
465 })),
466 };
467 let frame = PutRequest {
468 token: Some(PutToken {
469 value: b"opaque".to_vec(),
470 expires_at: None,
471 }),
472 chunk: Vec::new(),
473 };
474 assert!(matches!(
475 header.operation,
476 Some(put_header::Operation::PutIfVersion(_))
477 ));
478 assert!(frame.chunk.is_empty());
479
480 let operations = [
481 bulk_operation::Operation::Put(Default::default()),
482 bulk_operation::Operation::PutIfAbsent(Default::default()),
483 bulk_operation::Operation::PutIfVersion(BulkPutIfVersionRequest::default()),
484 bulk_operation::Operation::PutImmutable(Default::default()),
485 bulk_operation::Operation::Delete(DeleteRequest {
486 address: address.clone(),
487 ..Default::default()
488 }),
489 bulk_operation::Operation::DeleteIfVersion(DeleteIfVersionRequest {
490 address: address.clone(),
491 expected_version: 8,
492 ..Default::default()
493 }),
494 ];
495 assert_eq!(
496 operations
497 .into_iter()
498 .map(|operation| BulkOperation {
499 operation: Some(operation),
500 })
501 .count(),
502 6
503 );
504
505 let current_head_delete = DeleteIfVersionRequest {
506 address: address.clone(),
507 expected_version: 8,
508 ..Default::default()
509 };
510 let retained_version_delete = DeleteVersionRequest {
511 address,
512 version: 7,
513 ..Default::default()
514 };
515 assert_eq!(current_head_delete.expected_version, 8);
516 assert_eq!(retained_version_delete.version, 7);
517 let replaced_current = DeleteVersionResponse {
518 deleted: true,
519 replacement_tombstone_version: Some(9),
520 };
521 assert_eq!(replaced_current.replacement_tombstone_version, Some(9));
522 assert_eq!(
523 CreateBucketRequest::default().versioning,
524 ObjectVersioning::Unversioned as i32
525 );
526
527 let list_request = ListObjectsRequest {
528 tenant: "acme".into(),
529 bucket: "objects".into(),
530 prefix: "reports/".into(),
531 start_after: Some("reports/2025.json".into()),
532 limit: 100,
533 };
534 let list_response = ListObjectsResponse {
535 paths: vec!["reports/2026.json".into()],
536 has_more: false,
537 };
538 assert_eq!(
539 list_request.start_after.as_deref(),
540 Some("reports/2025.json")
541 );
542 assert_eq!(list_response.paths, vec!["reports/2026.json".to_owned()]);
543 }
544
545 #[test]
546 fn authorization_wire_types_preserve_scope_and_typed_unions() {
547 use super::v1::{
548 AnyUsersetSelector, AtLeastRevision, AuthzConsistency, AuthzScope, DirectRelation,
549 InheritRule, NamespaceDefinition, ObjectRef, Permission, PermissionRule,
550 PutSchemaRequest, RelationDefinition, SubjectSelector, Userset, authz_consistency,
551 object_ref, permission_rule, relation_definition, subject, subject_selector,
552 };
553
554 let account = ObjectRef {
555 namespace: "account".into(),
556 id: Some(object_ref::Id::OpaqueId("acme".into())),
557 };
558 let members = Userset {
559 object: Some(account),
560 relation: "member".into(),
561 };
562 let schema = NamespaceDefinition {
563 name: "ledger".into(),
564 relations: vec![
565 RelationDefinition {
566 name: "reader".into(),
567 kind: Some(relation_definition::Kind::Direct(DirectRelation {
568 allowed_subjects: vec![SubjectSelector {
569 selector: Some(subject_selector::Selector::AnyUserset(
570 AnyUsersetSelector {
571 namespace: "account".into(),
572 relation: "member".into(),
573 },
574 )),
575 }],
576 })),
577 },
578 RelationDefinition {
579 name: "read".into(),
580 kind: Some(relation_definition::Kind::Permission(Permission {
581 rules: vec![PermissionRule {
582 rule: Some(permission_rule::Rule::Inherit(InheritRule {
583 relation: "reader".into(),
584 })),
585 }],
586 })),
587 },
588 ],
589 };
590 let subject = super::v1::Subject {
591 kind: Some(subject::Kind::Userset(members)),
592 };
593 let publication = PutSchemaRequest {
594 schema_id: "acme".into(),
595 namespaces: vec![schema],
596 };
597 let consistency = AuthzConsistency {
598 requirement: Some(authz_consistency::Requirement::AtLeast(AtLeastRevision {
599 revision: 42,
600 })),
601 };
602 let system_scope = AuthzScope {
603 storage_tenant: "keldra-internal".into(),
604 realm: "_keldra/system".into(),
605 };
606
607 assert_eq!(publication.namespaces[0].relations.len(), 2);
608 assert!(matches!(subject.kind, Some(subject::Kind::Userset(_))));
609 assert!(matches!(
610 consistency.requirement,
611 Some(authz_consistency::Requirement::AtLeast(AtLeastRevision {
612 revision: 42
613 }))
614 ));
615 assert_eq!(system_scope.realm, "_keldra/system");
616 }
617
618 #[test]
619 fn tuple_mutation_and_batch_checks_share_request_scope() {
620 use super::v1::{
621 AuthzScope, CheckPermissionsRequest, MutateTuplesRequest, ObjectRef, PermissionCheck,
622 RelationTuple, Subject, TupleMutation, authz_consistency, object_ref, subject,
623 tuple_mutation,
624 };
625
626 let scope = AuthzScope {
627 storage_tenant: "acme".into(),
628 realm: "default".into(),
629 };
630 let ledger = ObjectRef {
631 namespace: "ledger".into(),
632 id: Some(object_ref::Id::OpaqueId("main".into())),
633 };
634 let alice = Subject {
635 kind: Some(subject::Kind::Object(ObjectRef {
636 namespace: "user".into(),
637 id: Some(object_ref::Id::OpaqueId("alice".into())),
638 })),
639 };
640 let tuple = RelationTuple {
641 object: Some(ledger.clone()),
642 relation: "reader".into(),
643 subject: Some(alice.clone()),
644 };
645 let mutation = MutateTuplesRequest {
646 scope: Some(scope.clone()),
647 operation_id: "grant-alice".into(),
648 expected_revision: Some(41),
649 mutations: vec![TupleMutation {
650 operation: Some(tuple_mutation::Operation::Add(tuple)),
651 }],
652 };
653 let checks = CheckPermissionsRequest {
654 scope: Some(scope),
655 checks: vec![PermissionCheck {
656 subject: Some(alice),
657 object: Some(ledger),
658 relation: "read".into(),
659 }],
660 consistency: Some(super::v1::AuthzConsistency {
661 requirement: Some(authz_consistency::Requirement::Exact(
662 super::v1::ExactRevision { revision: 42 },
663 )),
664 }),
665 };
666
667 assert_eq!(mutation.expected_revision, Some(41));
668 assert_eq!(mutation.mutations.len(), 1);
669 assert_eq!(checks.checks.len(), 1);
670 assert!(matches!(
671 checks
672 .consistency
673 .and_then(|consistency| consistency.requirement),
674 Some(authz_consistency::Requirement::Exact(_))
675 ));
676 }
677}