diff --git a/book/src/SUMMARY.md b/book/src/SUMMARY.md index 1594b4d..06a44b9 100644 --- a/book/src/SUMMARY.md +++ b/book/src/SUMMARY.md @@ -28,6 +28,7 @@ - [delete](./repo-delete.md) - [Hooks](./repo-hooks.md) - [Error Types](./repo-errors.md) + - [Scoped Repositories](./scoped-repositories.md) - [Transactions](./transactions.md) - [Connection Traits](./connection-traits.md) diff --git a/book/src/scoped-repositories.md b/book/src/scoped-repositories.md new file mode 100644 index 0000000..49526ad --- /dev/null +++ b/book/src/scoped-repositories.md @@ -0,0 +1,128 @@ +# Scoped Repositories + +Multi-tenant applications need data access that is impossible to misuse: a +query that *forgets* to filter by the tenant column is a silent cross-tenant +leak that compiles and passes all same-tenant tests. Scoped repositories make +that mistake unrepresentable — on a scoped repo, **every generated read +function requires a scope argument**, enforced by the compiler. + +## Declaring a scope column + +Mark exactly one column with `scope`: + +```rust,ignore +#[derive(EsRepo)] +#[es_repo( + entity = "Customer", + columns( + partner_id(ty = "PartnerId", scope), + email(ty = "String"), + ) +)] +pub struct Customers { + pool: PgPool, +} +``` + +`partner_id` remains an ordinary persisted column — it is populated on +`create` from the `NewCustomer`'s field like any other column. The `scope` +marker additionally generates an entity-named scope enum: + +```rust,ignore +pub enum CustomerScope { + All, // no filter — reads across all scopes + Only(PartnerId), // restricts every read to this scope value +} + +impl From for CustomerScope { /* => Only */ } +impl From<&PartnerId> for CustomerScope { /* => Only */ } +``` + +There is deliberately **no** `From>`: mapping `None` to +`All` would turn a stray `None` into silent all-scope access. All-scope reads +must be written explicitly — `CustomerScope::All` is greppable and auditable. + +## The scoped read surface + +Every generated read function gains a leading `scope: impl Into<{Entity}Scope>` +argument: + +```rust,ignore +customers.find_by_id(partner_id, id).await?; // Into => Only +customers.find_by_id(CustomerScope::All, id).await?; // explicit escape hatch +customers.maybe_find_by_email(partner_id, email).await?; +customers.find_all::(partner_id, &ids).await?; +customers.list_by_created_at(partner_id, args, direction).await?; +customers.list_for_filters(partner_id, filters, sort, args).await?; + +customers.find_by_id(id).await?; // does not exist — compile error +``` + +At runtime each function dispatches between two static, compile-time-checked +SQL variants: + +- `All` executes exactly the SQL an unscoped repo would. +- `Only(value)` executes a variant with an additional `partner_id = $n` + conjunct in every `WHERE` clause. + +Both arms are plain equality predicates — sargable against a scope-column-led +index (see below). Under `Only`, a row from another scope behaves exactly like +a missing row: `find_by_*` returns `NotFound`, `maybe_find_by_*` returns +`None`, `find_all` silently omits the id, and lists never contain the row. +**Missing and not-yours look identical.** + +## Writes are custody-guarded + +`create`, `create_all`, `update`, `update_all` and `delete` keep their +unscoped signatures. The reasoning: mutations operate on an entity value that +could only have been obtained through a scoped read (or built by domain logic +that stamped the scope column). Scope enforcement happens at the boundary that +turns ids and queries into entity data; once you hold the entity, custody of +the value is the guarantee. + +## Cursors carry no filter authority + +Pagination cursors are position markers only. Every page executes with the +scope conjunct in its own `WHERE` clause, so a tampered, fabricated, or +foreign cursor can only reposition pagination within the caller's own scoped +rows — it can never widen the result set, and cursor values are compared, not +dereferenced, so they cannot be used to probe for the existence of foreign +ids. Replaying a cursor minted under a different scope yields well-defined +(scoped) but position-shifted results. + +## Validation rules + +The macro rejects at compile time: + +- more than one `scope` column per repo +- an `Option` or `nullable`-annotated scope column (nullable scope columns + are not supported — every row must belong to exactly one scope) +- a `Forgettable` scope column +- `find_by`, `list_by` or `list_for` on the scope column itself — every read + is already filtered by it; per-scope listing *is* the ordinary scoped + `list_by_*(Only(value), ..)` +- `scope` on nested repos — children are custody-guarded via their (scoped) + parent + +The scope column also generates no `find_by_partner_id` accessors: the scope +argument replaces them. + +## Index requirements + +The `Only` arm adds a leading equality on the scope column to every read, so +composite indexes should lead with it: + +```sql +-- list_by_created_at under Only(p) +CREATE INDEX ON customers (partner_id, created_at DESC, id DESC); + +-- list_for_status_by_created_at under Only(p) +CREATE INDEX ON customers (partner_id, status, created_at DESC, id DESC); + +-- find_by_email under Only(p) +CREATE INDEX ON customers (partner_id, email); +``` + +Plain single-column indexes keep working (Postgres can still apply the scope +conjunct as an index qual or filter), but scope-led composites let the +paginated lists ride the index order with an early-exit `LIMIT`. diff --git a/es-entity-macros/src/repo/find_all_fn.rs b/es-entity-macros/src/repo/find_all_fn.rs index 77904df..663d579 100644 --- a/es-entity-macros/src/repo/find_all_fn.rs +++ b/es-entity-macros/src/repo/find_all_fn.rs @@ -2,7 +2,7 @@ use darling::ToTokens; use proc_macro2::TokenStream; use quote::{TokenStreamExt, quote}; -use super::options::*; +use super::{options::*, scope::ScopeInfo}; pub struct FindAllFn<'a> { prefix: Option<&'a syn::LitStr>, @@ -13,6 +13,7 @@ pub struct FindAllFn<'a> { any_nested: bool, post_hydrate_error: Option<&'a syn::Type>, forgettable_table_name: Option<&'a str>, + scope: Option>, #[cfg(feature = "instrument")] repo_name_snake: String, } @@ -28,6 +29,7 @@ impl<'a> From<&'a RepositoryOptions> for FindAllFn<'a> { any_nested: opts.any_nested(), post_hydrate_error: opts.post_hydrate_hook.as_ref().map(|h| &h.error), forgettable_table_name: opts.forgettable_table_name(), + scope: ScopeInfo::from_opts(opts), #[cfg(feature = "instrument")] repo_name_snake: opts.repo_name_snake_case(), } @@ -56,24 +58,48 @@ impl ToTokens for FindAllFn<'_> { quote! {} }; - let es_query_call = if let Some(prefix) = self.prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #query, - ids as &[#id], - ) + let make_es_query = |query: &str, extra_args: &TokenStream| -> TokenStream { + if let Some(prefix) = self.prefix { + quote! { + es_entity::es_query!( + tbl_prefix = #prefix, + #forgettable_tbl_arg + #query, + ids as &[#id], + #extra_args + ) + } + } else { + quote! { + es_entity::es_query!( + entity = #entity, + #forgettable_tbl_arg + #query, + ids as &[#id], + #extra_args + ) + } } + }; + let es_query_call = make_es_query(&query, "e! {}); + + let (scope_fn_arg, scope_fn_pass, scope_convert) = match &self.scope { + Some(scope) => (scope.fn_arg(), scope.fn_pass(), scope.convert()), + None => (quote! {}, quote! {}, quote! {}), + }; + let fetch_call = if let Some(scope) = &self.scope { + let scoped_query = format!( + "SELECT id FROM {} WHERE id = ANY($1) AND {}", + self.table_name, + scope.predicate(2), + ); + let scoped_es_query_call = make_es_query(&scoped_query, &scope.arg_tokens()); + scope.dispatch( + quote! { #es_query_call.fetch_n(op, ids.len()).await? }, + quote! { #scoped_es_query_call.fetch_n(op, ids.len()).await? }, + ) } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #query, - ids as &[#id], - ) - } + quote! { #es_query_call.fetch_n(op, ids.len()).await? } }; let op_param = if self.any_nested { @@ -107,18 +133,21 @@ impl ToTokens for FindAllFn<'_> { tokens.append_all(quote! { pub async fn find_all>( &self, + #scope_fn_arg ids: &[#id] ) -> Result, #query_error> { - self.find_all_in_op(#query_fn_get_op, ids).await + self.find_all_in_op(#query_fn_get_op, #scope_fn_pass ids).await } #instrument_attr pub async fn find_all_in_op #generics( &self, #op_param, + #scope_fn_arg ids: &[#id] ) -> Result, #query_error> { - let (entities, _) = #es_query_call.fetch_n(op, ids.len()).await?; + #scope_convert + let (entities, _) = #fetch_call; #post_hydrate_check Ok(entities.into_iter().map(|u| (u.id.clone(), Out::from(u))).collect()) } @@ -147,6 +176,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; diff --git a/es-entity-macros/src/repo/find_by_fn.rs b/es-entity-macros/src/repo/find_by_fn.rs index 5447b22..b85b863 100644 --- a/es-entity-macros/src/repo/find_by_fn.rs +++ b/es-entity-macros/src/repo/find_by_fn.rs @@ -3,7 +3,7 @@ use darling::ToTokens; use proc_macro2::{Span, TokenStream}; use quote::{TokenStreamExt, quote}; -use super::options::*; +use super::{options::*, scope::ScopeInfo}; pub struct FindByFn<'a> { prefix: Option<&'a syn::LitStr>, @@ -17,6 +17,7 @@ pub struct FindByFn<'a> { any_nested: bool, post_hydrate_error: Option<&'a syn::Type>, forgettable_table_name: Option<&'a str>, + scope: Option>, #[cfg(feature = "instrument")] repo_name_snake: String, } @@ -35,6 +36,7 @@ impl<'a> FindByFn<'a> { any_nested: opts.any_nested(), post_hydrate_error: opts.post_hydrate_hook.as_ref().map(|h| &h.error), forgettable_table_name: opts.forgettable_table_name(), + scope: ScopeInfo::from_opts(opts), #[cfg(feature = "instrument")] repo_name_snake: opts.repo_name_snake_case(), } @@ -51,6 +53,11 @@ impl ToTokens for FindByFn<'_> { let query_fn_op_traits = RepositoryOptions::query_fn_op_traits(self.any_nested); let query_fn_get_op = RepositoryOptions::query_fn_get_op(self.any_nested); + let (scope_fn_arg, scope_fn_pass, scope_convert) = match &self.scope { + Some(scope) => (scope.fn_arg(), scope.fn_pass(), scope.convert()), + None => (quote! {}, quote! {}, quote! {}), + }; + for maybe in ["", "maybe_"] { let error = if maybe.is_empty() { &self.find_error @@ -107,30 +114,56 @@ impl ToTokens for FindByFn<'_> { quote! {} }; - let es_query_call = if let Some(prefix) = self.prefix { - quote! { - es_entity::es_query!( - tbl_prefix = #prefix, - #forgettable_tbl_arg - #query, - #column_name as &#column_type, - ) - } - } else { - quote! { - es_entity::es_query!( - entity = #entity, - #forgettable_tbl_arg - #query, - #column_name as &#column_type, - ) + let make_es_query = |query: &str, extra_args: &TokenStream| -> TokenStream { + if let Some(prefix) = self.prefix { + quote! { + es_entity::es_query!( + tbl_prefix = #prefix, + #forgettable_tbl_arg + #query, + #column_name as &#column_type, + #extra_args + ) + } + } else { + quote! { + es_entity::es_query!( + entity = #entity, + #forgettable_tbl_arg + #query, + #column_name as &#column_type, + #extra_args + ) + } } }; + let es_query_call = make_es_query(&query, "e! {}); - let fetch_optional_call = if delete == DeleteOption::Soft && self.any_nested { - quote! { #es_query_call.fetch_optional_include_deleted(op).await? } + let fetch_method = if delete == DeleteOption::Soft && self.any_nested { + quote! { fetch_optional_include_deleted } + } else { + quote! { fetch_optional } + }; + let fetch_optional_call = if let Some(scope) = &self.scope { + let scoped_query = format!( + r#"SELECT id FROM {} WHERE {} {} $1 AND {}{}"#, + self.table_name, + column_name, + filter_op, + scope.predicate(2), + if delete == DeleteOption::No { + self.delete.not_deleted_condition() + } else { + "" + } + ); + let scoped_es_query_call = make_es_query(&scoped_query, &scope.arg_tokens()); + scope.dispatch( + quote! { #es_query_call.#fetch_method(op).await? }, + quote! { #scoped_es_query_call.#fetch_method(op).await? }, + ) } else { - quote! { #es_query_call.fetch_optional(op).await? } + quote! { #es_query_call.#fetch_method(op).await? } }; let fetch_and_validate = if maybe.is_empty() { @@ -206,21 +239,24 @@ impl ToTokens for FindByFn<'_> { tokens.append_all(quote! { pub async fn #fn_name( &self, + #scope_fn_arg #column_name: #impl_expr ) -> Result<#result_type, #error> { - self.#fn_in_op(#query_fn_get_op, #column_name).await + self.#fn_in_op(#query_fn_get_op, #scope_fn_pass #column_name).await } #instrument_attr_in_op pub async fn #fn_in_op #query_fn_generics( &self, #query_fn_op_arg, + #scope_fn_arg #column_name: #impl_expr ) -> Result<#result_type, #error> where OP: #query_fn_op_traits { let __result: Result<#result_type, #error> = async { + #scope_convert let #column_name = #column_name.#access_expr; #record_field #fetch_and_validate @@ -262,6 +298,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -359,6 +396,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -453,6 +491,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -547,6 +586,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -576,6 +616,7 @@ mod tests { any_nested: true, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -607,6 +648,7 @@ mod tests { any_nested: true, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; diff --git a/es-entity-macros/src/repo/list_by_fn.rs b/es-entity-macros/src/repo/list_by_fn.rs index e1cbf5c..c0ac0f7 100644 --- a/es-entity-macros/src/repo/list_by_fn.rs +++ b/es-entity-macros/src/repo/list_by_fn.rs @@ -3,7 +3,7 @@ use darling::ToTokens; use proc_macro2::{Span, TokenStream}; use quote::{TokenStreamExt, quote}; -use super::options::*; +use super::{options::*, scope::ScopeInfo}; /// Cursor pagination states that each get their own SQL text, so that no /// variant needs the non-sargable `COALESCE(..., $ IS NULL)` catch-all. @@ -443,6 +443,7 @@ pub struct ListByFn<'a> { any_nested: bool, post_hydrate_error: Option<&'a syn::Type>, forgettable_table_name: Option<&'a str>, + scope: Option>, #[cfg(feature = "instrument")] repo_name_snake: String, } @@ -461,6 +462,7 @@ impl<'a> ListByFn<'a> { any_nested: opts.any_nested(), post_hydrate_error: opts.post_hydrate_hook.as_ref().map(|h| &h.error), forgettable_table_name: opts.forgettable_table_name(), + scope: ScopeInfo::from_opts(opts), #[cfg(feature = "instrument")] repo_name_snake: opts.repo_name_snake_case(), } @@ -538,46 +540,84 @@ impl ToTokens for ListByFn<'_> { } }; - let mut query_arms = TokenStream::new(); - for state in cursor.cursor_states() { - for ascending in [true, false] { - let mut conditions: Vec = Vec::new(); - if let Some(condition) = cursor.condition_for_state(*state, 0, ascending) { - conditions.push(format!("({condition})")); - } - if delete == DeleteOption::No - && let Some(not_deleted) = not_deleted_predicate(self.delete) - { - conditions.push(not_deleted); + let build_query_arms = |scope: Option<&ScopeInfo>| -> TokenStream { + let mut query_arms = TokenStream::new(); + let offset = if scope.is_some() { 1 } else { 0 }; + for state in cursor.cursor_states() { + for ascending in [true, false] { + let mut conditions: Vec = Vec::new(); + if let Some(scope) = scope { + conditions.push(scope.predicate(1)); + } + if let Some(condition) = + cursor.condition_for_state(*state, offset, ascending) + { + conditions.push(format!("({condition})")); + } + if delete == DeleteOption::No + && let Some(not_deleted) = not_deleted_predicate(self.delete) + { + conditions.push(not_deleted); + } + let query = assemble_select( + &select_columns, + self.table_name, + &conditions, + &cursor.order_by(ascending), + offset + 1, + ); + let cursor_args = cursor.cursor_arg_tokens_for_state(*state); + let scope_args = scope.map(|s| s.arg_tokens()).unwrap_or_default(); + let args = quote! { + #scope_args + (first + 1) as i64, + #cursor_args + }; + let es_query_call = make_es_query(&query, &args); + let direction_pattern = if ascending { + quote! { es_entity::ListDirection::Ascending } + } else { + quote! { es_entity::ListDirection::Descending } + }; + let state_pattern = cursor.state_pattern_elems(*state); + query_arms.append_all(quote! { + (#direction_pattern, #(#state_pattern),*) => { + #es_query_call.fetch_n(op, first).await? + }, + }); } - let query = assemble_select( - &select_columns, - self.table_name, - &conditions, - &cursor.order_by(ascending), - 1, - ); - let cursor_args = cursor.cursor_arg_tokens_for_state(*state); - let args = quote! { - (first + 1) as i64, - #cursor_args - }; - let es_query_call = make_es_query(&query, &args); - let direction_pattern = if ascending { - quote! { es_entity::ListDirection::Ascending } - } else { - quote! { es_entity::ListDirection::Descending } - }; - let state_pattern = cursor.state_pattern_elems(*state); - query_arms.append_all(quote! { - (#direction_pattern, #(#state_pattern),*) => { - #es_query_call.fetch_n(op, first).await? - }, - }); } - } + query_arms + }; + let query_arms = build_query_arms(None); let cursor_state_scrutinee = cursor.state_scrutinee_elems(); + let (scope_fn_arg, scope_fn_pass, scope_convert) = match &self.scope { + Some(scope) => (scope.fn_arg(), scope.fn_pass(), scope.convert()), + None => (quote! {}, quote! {}, quote! {}), + }; + let match_expr = if let Some(scope) = &self.scope { + let scoped_query_arms = build_query_arms(Some(scope)); + scope.dispatch( + quote! { + match (direction, #(#cursor_state_scrutinee),*) { + #query_arms + } + }, + quote! { + match (direction, #(#cursor_state_scrutinee),*) { + #scoped_query_arms + } + }, + ) + } else { + quote! { + match (direction, #(#cursor_state_scrutinee),*) { + #query_arms + } + } + }; + #[cfg(feature = "instrument")] let ( instrument_attr, @@ -637,16 +677,18 @@ impl ToTokens for ListByFn<'_> { tokens.append_all(quote! { pub async fn #fn_name( &self, + #scope_fn_arg cursor: es_entity::PaginatedQueryArgs<#cursor_mod::#cursor_ident>, direction: es_entity::ListDirection, ) -> Result, #query_error> { - self.#fn_in_op(#query_fn_get_op, cursor, direction).await + self.#fn_in_op(#query_fn_get_op, #scope_fn_pass cursor, direction).await } #instrument_attr pub async fn #fn_in_op #query_fn_generics( &self, #query_fn_op_arg, + #scope_fn_arg cursor: es_entity::PaginatedQueryArgs<#cursor_mod::#cursor_ident>, direction: es_entity::ListDirection, ) -> Result, #query_error> @@ -654,13 +696,12 @@ impl ToTokens for ListByFn<'_> { OP: #query_fn_op_traits { let __result: Result, #query_error> = async { + #scope_convert #extract_has_cursor #destructure_tokens #record_fields - let (entities, has_next_page) = match (direction, #(#cursor_state_scrutinee),*) { - #query_arms - }; + let (entities, has_next_page) = #match_expr; #post_hydrate_check #record_results @@ -787,6 +828,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -896,6 +938,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -930,6 +973,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -1044,6 +1088,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -1190,6 +1235,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; diff --git a/es-entity-macros/src/repo/list_for_filters_fn.rs b/es-entity-macros/src/repo/list_for_filters_fn.rs index fa2b9bb..a2b2348 100644 --- a/es-entity-macros/src/repo/list_for_filters_fn.rs +++ b/es-entity-macros/src/repo/list_for_filters_fn.rs @@ -7,6 +7,7 @@ use super::{ combo_cursor::ComboCursor, list_by_fn::{CursorStruct, assemble_select, not_deleted_predicate}, options::*, + scope::ScopeInfo, }; /// Runtime `Some`-ness state of one filter column. Each state that reaches @@ -174,6 +175,7 @@ pub struct ListForFiltersFn<'a> { any_nested: bool, post_hydrate_error: Option<&'a syn::Type>, forgettable_table_name: Option<&'a str>, + scope: Option>, #[cfg(feature = "instrument")] repo_name_snake: String, } @@ -200,6 +202,7 @@ impl<'a> ListForFiltersFn<'a> { any_nested: opts.any_nested(), post_hydrate_error: opts.post_hydrate_hook.as_ref().map(|h| &h.error), forgettable_table_name: opts.forgettable_table_name(), + scope: ScopeInfo::from_opts(opts), #[cfg(feature = "instrument")] repo_name_snake: opts.repo_name_snake_case(), } @@ -267,13 +270,19 @@ impl<'a> ListForFiltersFn<'a> { let by_col_name = by_col.name(); let delete_postfix = delete.include_deletion_fn_postfix(); + let scope_pass = if self.scope.is_some() { + quote! { __scope, } + } else { + quote! {} + }; + let list_by_fn = syn::Ident::new( &format!("list_by_{}{}", by_col_name, delete_postfix), Span::call_site(), ); if self.for_columns.is_empty() { - return quote! { self.#list_by_fn(query, direction).await? }; + return quote! { self.#list_by_fn(#scope_pass query, direction).await? }; } let all_none_checks: Vec<_> = self @@ -317,13 +326,13 @@ impl<'a> ListForFiltersFn<'a> { if others_none.is_empty() { quote! { else { - self.#fn_name(filters.#for_col_name.unwrap(), query, direction).await? + self.#fn_name(#scope_pass filters.#for_col_name.unwrap(), query, direction).await? } } } else { quote! { else if #(#others_none)&&* { - self.#fn_name(filters.#for_col_name.unwrap(), query, direction).await? + self.#fn_name(#scope_pass filters.#for_col_name.unwrap(), query, direction).await? } } } @@ -343,7 +352,7 @@ impl<'a> ListForFiltersFn<'a> { ); quote! { else { - self.#list_for_filters_fn(filters, query, direction).await? + self.#list_for_filters_fn(#scope_pass filters, query, direction).await? } } } else { @@ -352,7 +361,7 @@ impl<'a> ListForFiltersFn<'a> { quote! { if #(#all_none_checks)&&* { - self.#list_by_fn(query, direction).await? + self.#list_by_fn(#scope_pass query, direction).await? } #single_filter_branches #multi_filter_fallback @@ -429,60 +438,71 @@ impl<'a> ListForFiltersFn<'a> { }) .collect(); - // Generate WHERE clause fragments - let mut param_idx = 1u32; - let where_fragments: Vec = self - .for_columns - .iter() - .map(|col| FiltersStruct::where_clause_fragment(col, &mut param_idx)) - .collect(); + // Generate the catch-all fallback query (COALESCE-style, correctness + // fallback for filter combinations above the specialization cap). + // Parameterized over the scope: the scoped variant binds the scope + // column at `$1` and shifts every other parameter by one. + let build_fallback = |scope: Option<&ScopeInfo>| -> (String, String, TokenStream) { + let scope_offset: u32 = if scope.is_some() { 1 } else { 0 }; + let mut param_idx = 1u32 + scope_offset; + let where_fragments: Vec = self + .for_columns + .iter() + .map(|col| FiltersStruct::where_clause_fragment(col, &mut param_idx)) + .collect(); - let filter_where = if where_fragments.is_empty() { - String::new() - } else { - format!("{} AND ", where_fragments.join(" AND ")) - }; + let mut filter_where = if where_fragments.is_empty() { + String::new() + } else { + format!("{} AND ", where_fragments.join(" AND ")) + }; + if let Some(scope) = scope { + filter_where = format!("{} AND {}", scope.predicate(1), filter_where); + } - // Generate filter arg bindings for es_query! - let filter_arg_bindings: TokenStream = self - .for_columns - .iter() - .map(|col| FiltersStruct::filter_arg_tokens(col)) - .collect(); + let filter_arg_bindings: TokenStream = self + .for_columns + .iter() + .map(|col| FiltersStruct::filter_arg_tokens(col)) + .collect(); + let scope_args = scope.map(|s| s.arg_tokens()).unwrap_or_default(); + let fallback_arg_tokens = quote! { + #scope_args + #filter_arg_bindings + #cursor_arg_tokens + }; - let fallback_arg_tokens = quote! { - #filter_arg_bindings - #cursor_arg_tokens + let asc_query = format!( + r#"SELECT {} FROM {} WHERE {}({}){} ORDER BY {} LIMIT ${}"#, + select_columns, + self.table_name, + filter_where, + cursor_struct.condition(n_filters + scope_offset, true), + if delete == DeleteOption::No { + self.delete.not_deleted_condition() + } else { + "" + }, + cursor_struct.order_by(true), + n_filters + scope_offset + 1, + ); + let desc_query = format!( + r#"SELECT {} FROM {} WHERE {}({}){} ORDER BY {} LIMIT ${}"#, + select_columns, + self.table_name, + filter_where, + cursor_struct.condition(n_filters + scope_offset, false), + if delete == DeleteOption::No { + self.delete.not_deleted_condition() + } else { + "" + }, + cursor_struct.order_by(false), + n_filters + scope_offset + 1, + ); + (asc_query, desc_query, fallback_arg_tokens) }; - - let asc_query = format!( - r#"SELECT {} FROM {} WHERE {}({}){} ORDER BY {} LIMIT ${}"#, - select_columns, - self.table_name, - filter_where, - cursor_struct.condition(n_filters, true), - if delete == DeleteOption::No { - self.delete.not_deleted_condition() - } else { - "" - }, - cursor_struct.order_by(true), - n_filters + 1, - ); - let desc_query = format!( - r#"SELECT {} FROM {} WHERE {}({}){} ORDER BY {} LIMIT ${}"#, - select_columns, - self.table_name, - filter_where, - cursor_struct.condition(n_filters, false), - if delete == DeleteOption::No { - self.delete.not_deleted_condition() - } else { - "" - }, - cursor_struct.order_by(false), - n_filters + 1, - ); + let (asc_query, desc_query, fallback_arg_tokens) = build_fallback(None); let forgettable_tbl_arg = if let Some(tbl) = self.forgettable_table_name { quote! { forgettable_tbl = #tbl, } @@ -516,90 +536,103 @@ impl<'a> ListForFiltersFn<'a> { // combination x cursor state x direction). Every present filter // compiles to a sargable `col = $k` (or `col IS NULL`) predicate and // the cursor predicate is either omitted (page 1) or a bare row - // comparison. - let mut asc_arms = TokenStream::new(); - let mut desc_arms = TokenStream::new(); - let mut all_combos_specialized = true; - for combo in filter_state_combos(&self.for_columns) { - if !self.is_specialized_combo(&combo) { - all_combos_specialized = false; - continue; - } - let filter_patterns: Vec = self - .for_columns - .iter() - .zip(combo.iter()) - .flat_map(|(col, state)| Self::filter_pattern_elems(col, *state)) - .collect(); - - let mut filter_conditions: Vec = Vec::new(); - let mut filter_args = TokenStream::new(); - let mut param_idx = 1u32; - for (col, state) in self.for_columns.iter().zip(combo.iter()) { - match state { - FilterState::Absent => {} - FilterState::Present => { - filter_conditions.push(format!("{} = ${}", col.name(), param_idx)); - param_idx += 1; - filter_args.append_all(FiltersStruct::filter_arg_tokens(col)); + // comparison. Parameterized over the scope: the scoped variant binds + // the scope column at `$1` and shifts every other parameter by one. + let build_specialized_arms = + |scope: Option<&ScopeInfo>| -> (TokenStream, TokenStream, bool) { + let mut asc_arms = TokenStream::new(); + let mut desc_arms = TokenStream::new(); + let mut all_combos_specialized = true; + for combo in filter_state_combos(&self.for_columns) { + if !self.is_specialized_combo(&combo) { + all_combos_specialized = false; + continue; } - FilterState::PresentNull => { - filter_conditions.push(format!("{} IS NULL", col.name())); - } - FilterState::PresentValue => { - filter_conditions.push(format!("{} = ${}", col.name(), param_idx)); + let filter_patterns: Vec = self + .for_columns + .iter() + .zip(combo.iter()) + .flat_map(|(col, state)| Self::filter_pattern_elems(col, *state)) + .collect(); + + let mut filter_conditions: Vec = Vec::new(); + let mut filter_args = TokenStream::new(); + let mut param_idx = 1u32; + if let Some(scope) = scope { + filter_conditions.push(scope.predicate(1)); + filter_args.append_all(scope.arg_tokens()); param_idx += 1; - filter_args.append_all(FiltersStruct::filter_value_arg_tokens(col)); - } - } - } - - for cursor_state in cursor_struct.cursor_states() { - let cursor_patterns = cursor_struct.state_pattern_elems(*cursor_state); - let pattern = quote! { (#(#filter_patterns,)* #(#cursor_patterns,)*) }; - let cursor_args = cursor_struct.cursor_arg_tokens_for_state(*cursor_state); - let args = quote! { - #filter_args - (first + 1) as i64, - #cursor_args - }; - - for ascending in [true, false] { - let mut conditions = filter_conditions.clone(); - if let Some(condition) = - cursor_struct.condition_for_state(*cursor_state, param_idx - 1, ascending) - { - conditions.push(format!("({condition})")); } - if delete == DeleteOption::No - && let Some(not_deleted) = not_deleted_predicate(self.delete) - { - conditions.push(not_deleted); + for (col, state) in self.for_columns.iter().zip(combo.iter()) { + match state { + FilterState::Absent => {} + FilterState::Present => { + filter_conditions.push(format!("{} = ${}", col.name(), param_idx)); + param_idx += 1; + filter_args.append_all(FiltersStruct::filter_arg_tokens(col)); + } + FilterState::PresentNull => { + filter_conditions.push(format!("{} IS NULL", col.name())); + } + FilterState::PresentValue => { + filter_conditions.push(format!("{} = ${}", col.name(), param_idx)); + param_idx += 1; + filter_args.append_all(FiltersStruct::filter_value_arg_tokens(col)); + } + } } - let query = assemble_select( - &select_columns, - self.table_name, - &conditions, - &cursor_struct.order_by(ascending), - param_idx, - ); - let es_query_call = make_es_query(&query, &args); - if ascending { - asc_arms.append_all(quote! { - #pattern => { - #es_query_call.fetch_n(op, first).await? - }, - }); - } else { - desc_arms.append_all(quote! { - #pattern => { - #es_query_call.fetch_n(op, first).await? - }, - }); + + for cursor_state in cursor_struct.cursor_states() { + let cursor_patterns = cursor_struct.state_pattern_elems(*cursor_state); + let pattern = quote! { (#(#filter_patterns,)* #(#cursor_patterns,)*) }; + let cursor_args = cursor_struct.cursor_arg_tokens_for_state(*cursor_state); + let args = quote! { + #filter_args + (first + 1) as i64, + #cursor_args + }; + + for ascending in [true, false] { + let mut conditions = filter_conditions.clone(); + if let Some(condition) = cursor_struct.condition_for_state( + *cursor_state, + param_idx - 1, + ascending, + ) { + conditions.push(format!("({condition})")); + } + if delete == DeleteOption::No + && let Some(not_deleted) = not_deleted_predicate(self.delete) + { + conditions.push(not_deleted); + } + let query = assemble_select( + &select_columns, + self.table_name, + &conditions, + &cursor_struct.order_by(ascending), + param_idx, + ); + let es_query_call = make_es_query(&query, &args); + if ascending { + asc_arms.append_all(quote! { + #pattern => { + #es_query_call.fetch_n(op, first).await? + }, + }); + } else { + desc_arms.append_all(quote! { + #pattern => { + #es_query_call.fetch_n(op, first).await? + }, + }); + } + } } } - } - } + (asc_arms, desc_arms, all_combos_specialized) + }; + let (asc_arms, desc_arms, all_combos_specialized) = build_specialized_arms(None); let scrutinee_elems: Vec = self .filter_scrutinee_elems() @@ -610,15 +643,73 @@ impl<'a> ListForFiltersFn<'a> { // When every filter combination is specialized the explicit arms // already cover the entire pattern space, so no wildcard fallback arm // (nor its catch-all COALESCE queries) is emitted. - let (asc_fallback_arm, desc_fallback_arm) = if all_combos_specialized { - (quote! {}, quote! {}) - } else { - let asc_call = make_es_query(&asc_query, &fallback_arg_tokens); - let desc_call = make_es_query(&desc_query, &fallback_arg_tokens); - ( - quote! { _ => #asc_call.fetch_n(op, first).await?, }, - quote! { _ => #desc_call.fetch_n(op, first).await?, }, + let build_fallback_arms = |asc_query: &str, + desc_query: &str, + args: &TokenStream, + all_specialized: bool| + -> (TokenStream, TokenStream) { + if all_specialized { + (quote! {}, quote! {}) + } else { + let asc_call = make_es_query(asc_query, args); + let desc_call = make_es_query(desc_query, args); + ( + quote! { _ => #asc_call.fetch_n(op, first).await?, }, + quote! { _ => #desc_call.fetch_n(op, first).await?, }, + ) + } + }; + let (asc_fallback_arm, desc_fallback_arm) = build_fallback_arms( + &asc_query, + &desc_query, + &fallback_arg_tokens, + all_combos_specialized, + ); + + let direction_match = |asc_arms: &TokenStream, + asc_fallback: &TokenStream, + desc_arms: &TokenStream, + desc_fallback: &TokenStream| + -> TokenStream { + quote! { + match direction { + es_entity::ListDirection::Ascending => match (#(#scrutinee_elems,)*) { + #asc_arms + #asc_fallback + }, + es_entity::ListDirection::Descending => match (#(#scrutinee_elems,)*) { + #desc_arms + #desc_fallback + } + } + } + }; + let (scope_fn_arg, scope_fn_pass, scope_convert) = match &self.scope { + Some(scope) => (scope.fn_arg(), scope.fn_pass(), scope.convert()), + None => (quote! {}, quote! {}, quote! {}), + }; + let match_expr = if let Some(scope) = &self.scope { + let (scoped_asc_arms, scoped_desc_arms, scoped_all_specialized) = + build_specialized_arms(Some(scope)); + let (scoped_asc_query, scoped_desc_query, scoped_fallback_args) = + build_fallback(Some(scope)); + let (scoped_asc_fallback, scoped_desc_fallback) = build_fallback_arms( + &scoped_asc_query, + &scoped_desc_query, + &scoped_fallback_args, + scoped_all_specialized, + ); + scope.dispatch( + direction_match(&asc_arms, &asc_fallback_arm, &desc_arms, &desc_fallback_arm), + direction_match( + &scoped_asc_arms, + &scoped_asc_fallback, + &scoped_desc_arms, + &scoped_desc_fallback, + ), ) + } else { + direction_match(&asc_arms, &asc_fallback_arm, &desc_arms, &desc_fallback_arm) }; #[cfg(feature = "instrument")] @@ -669,17 +760,19 @@ impl<'a> ListForFiltersFn<'a> { quote! { pub async fn #fn_name( &self, + #scope_fn_arg filters: #filters_ident, cursor: es_entity::PaginatedQueryArgs<#cursor_mod::#cursor_ident>, direction: es_entity::ListDirection, ) -> Result, #error> { - self.#fn_in_op(#query_fn_get_op, filters, cursor, direction).await + self.#fn_in_op(#query_fn_get_op, #scope_fn_pass filters, cursor, direction).await } #instrument_attr pub async fn #fn_in_op #query_fn_generics( &self, #query_fn_op_arg, + #scope_fn_arg filters: #filters_ident, cursor: es_entity::PaginatedQueryArgs<#cursor_mod::#cursor_ident>, direction: es_entity::ListDirection, @@ -688,21 +781,13 @@ impl<'a> ListForFiltersFn<'a> { OP: #query_fn_op_traits { let __result: Result, #error> = async { + #scope_convert #extract_has_cursor #destructure_filters #destructure_tokens #record_fields - let (entities, has_next_page) = match direction { - es_entity::ListDirection::Ascending => match (#(#scrutinee_elems,)*) { - #asc_arms - #asc_fallback_arm - }, - es_entity::ListDirection::Descending => match (#(#scrutinee_elems,)*) { - #desc_arms - #desc_fallback_arm - } - }; + let (entities, has_next_page) = #match_expr; #post_hydrate_check #record_results @@ -733,6 +818,11 @@ impl ToTokens for ListForFiltersFn<'_> { let error = &self.query_error; let cursor_mod = &self.cursor_mod; + let (scope_fn_arg, scope_convert) = match &self.scope { + Some(scope) => (scope.fn_arg(), scope.convert()), + None => (quote! {}, quote! {}), + }; + for delete in [DeleteOption::No, DeleteOption::Soft] { // Generate per-sort-column functions let by_fns: TokenStream = self @@ -836,12 +926,14 @@ impl ToTokens for ListForFiltersFn<'_> { #instrument_attr pub async fn #fn_name( &self, + #scope_fn_arg filters: #filters_name, sort: es_entity::Sort<#sort_by_name>, cursor: es_entity::PaginatedQueryArgs<#cursor_mod::#cursor_ident>, ) -> Result, #error> { let __result: Result, #error> = async { + #scope_convert #extract_has_cursor let es_entity::Sort { by, direction } = sort; let es_entity::PaginatedQueryArgs { first, after } = cursor; @@ -950,6 +1042,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -1270,6 +1363,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -1338,6 +1432,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -1423,6 +1518,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -1506,6 +1602,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -1595,6 +1692,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; diff --git a/es-entity-macros/src/repo/list_for_fn.rs b/es-entity-macros/src/repo/list_for_fn.rs index 675de4c..2776325 100644 --- a/es-entity-macros/src/repo/list_for_fn.rs +++ b/es-entity-macros/src/repo/list_for_fn.rs @@ -5,6 +5,7 @@ use quote::{TokenStreamExt, quote}; use super::{ list_by_fn::{CursorStruct, assemble_select, not_deleted_predicate}, options::*, + scope::ScopeInfo, }; pub struct ListForFn<'a> { @@ -20,6 +21,7 @@ pub struct ListForFn<'a> { any_nested: bool, post_hydrate_error: Option<&'a syn::Type>, forgettable_table_name: Option<&'a str>, + scope: Option>, #[cfg(feature = "instrument")] repo_name_snake: String, } @@ -39,6 +41,7 @@ impl<'a> ListForFn<'a> { any_nested: opts.any_nested(), post_hydrate_error: opts.post_hydrate_hook.as_ref().map(|h| &h.error), forgettable_table_name: opts.forgettable_table_name(), + scope: ScopeInfo::from_opts(opts), #[cfg(feature = "instrument")] repo_name_snake: opts.repo_name_snake_case(), } @@ -132,48 +135,86 @@ impl ToTokens for ListForFn<'_> { } }; - let mut query_arms = TokenStream::new(); - for state in cursor.cursor_states() { - for ascending in [true, false] { - let mut conditions: Vec = - vec![format!("({for_column_name} {filter_op} $1)")]; - if let Some(condition) = cursor.condition_for_state(*state, 1, ascending) { - conditions.push(format!("({condition})")); - } - if delete == DeleteOption::No - && let Some(not_deleted) = not_deleted_predicate(self.delete) - { - conditions.push(not_deleted); + let build_query_arms = |scope: Option<&ScopeInfo>| -> TokenStream { + let mut query_arms = TokenStream::new(); + let offset = if scope.is_some() { 2 } else { 1 }; + for state in cursor.cursor_states() { + for ascending in [true, false] { + let mut conditions: Vec = + vec![format!("({for_column_name} {filter_op} $1)")]; + if let Some(scope) = scope { + conditions.push(scope.predicate(2)); + } + if let Some(condition) = + cursor.condition_for_state(*state, offset, ascending) + { + conditions.push(format!("({condition})")); + } + if delete == DeleteOption::No + && let Some(not_deleted) = not_deleted_predicate(self.delete) + { + conditions.push(not_deleted); + } + let query = assemble_select( + &select_columns, + self.table_name, + &conditions, + &cursor.order_by(ascending), + offset + 1, + ); + let cursor_args = cursor.cursor_arg_tokens_for_state(*state); + let scope_args = scope.map(|s| s.arg_tokens()).unwrap_or_default(); + let args = quote! { + #filter_arg_name as &#for_column_type, + #scope_args + (first + 1) as i64, + #cursor_args + }; + let es_query_call = make_es_query(&query, &args); + let direction_pattern = if ascending { + quote! { es_entity::ListDirection::Ascending } + } else { + quote! { es_entity::ListDirection::Descending } + }; + let state_pattern = cursor.state_pattern_elems(*state); + query_arms.append_all(quote! { + (#direction_pattern, #(#state_pattern),*) => { + #es_query_call.fetch_n(op, first).await? + }, + }); } - let query = assemble_select( - &select_columns, - self.table_name, - &conditions, - &cursor.order_by(ascending), - 2, - ); - let cursor_args = cursor.cursor_arg_tokens_for_state(*state); - let args = quote! { - #filter_arg_name as &#for_column_type, - (first + 1) as i64, - #cursor_args - }; - let es_query_call = make_es_query(&query, &args); - let direction_pattern = if ascending { - quote! { es_entity::ListDirection::Ascending } - } else { - quote! { es_entity::ListDirection::Descending } - }; - let state_pattern = cursor.state_pattern_elems(*state); - query_arms.append_all(quote! { - (#direction_pattern, #(#state_pattern),*) => { - #es_query_call.fetch_n(op, first).await? - }, - }); } - } + query_arms + }; + let query_arms = build_query_arms(None); let cursor_state_scrutinee = cursor.state_scrutinee_elems(); + let (scope_fn_arg, scope_fn_pass, scope_convert) = match &self.scope { + Some(scope) => (scope.fn_arg(), scope.fn_pass(), scope.convert()), + None => (quote! {}, quote! {}, quote! {}), + }; + let match_expr = if let Some(scope) = &self.scope { + let scoped_query_arms = build_query_arms(Some(scope)); + scope.dispatch( + quote! { + match (direction, #(#cursor_state_scrutinee),*) { + #query_arms + } + }, + quote! { + match (direction, #(#cursor_state_scrutinee),*) { + #scoped_query_arms + } + }, + ) + } else { + quote! { + match (direction, #(#cursor_state_scrutinee),*) { + #query_arms + } + } + }; + #[cfg(feature = "instrument")] let ( instrument_attr, @@ -240,17 +281,19 @@ impl ToTokens for ListForFn<'_> { tokens.append_all(quote! { pub async fn #fn_name( &self, + #scope_fn_arg #filter_arg_name: #for_impl_expr, cursor: es_entity::PaginatedQueryArgs<#cursor_mod::#cursor_ident>, direction: es_entity::ListDirection, ) -> Result, #error> { - self.#fn_in_op(#query_fn_get_op, #filter_arg_name, cursor, direction).await + self.#fn_in_op(#query_fn_get_op, #scope_fn_pass #filter_arg_name, cursor, direction).await } #instrument_attr pub async fn #fn_in_op #query_fn_generics( &self, #query_fn_op_arg, + #scope_fn_arg #filter_arg_name: #for_impl_expr, cursor: es_entity::PaginatedQueryArgs<#cursor_mod::#cursor_ident>, direction: es_entity::ListDirection, @@ -259,14 +302,13 @@ impl ToTokens for ListForFn<'_> { OP: #query_fn_op_traits { let __result: Result, #error> = async { + #scope_convert #extract_has_cursor let #filter_arg_name = #filter_arg_name.#for_access_expr; #destructure_tokens #record_fields - let (entities, has_next_page) = match (direction, #(#cursor_state_scrutinee),*) { - #query_arms - }; + let (entities, has_next_page) = #match_expr; #post_hydrate_check #record_results @@ -323,6 +365,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; @@ -442,6 +485,7 @@ mod tests { any_nested: false, post_hydrate_error: None, forgettable_table_name: None, + scope: None, #[cfg(feature = "instrument")] repo_name_snake: "test_repo".to_string(), }; diff --git a/es-entity-macros/src/repo/mod.rs b/es-entity-macros/src/repo/mod.rs index 415ddcd..a2be47c 100644 --- a/es-entity-macros/src/repo/mod.rs +++ b/es-entity-macros/src/repo/mod.rs @@ -17,6 +17,7 @@ mod persist_events_fn; mod populate_nested; mod post_hydrate_hook; mod post_persist_hook; +mod scope; mod update_all_fn; mod update_fn; @@ -29,6 +30,7 @@ use options::RepositoryOptions; pub fn derive(ast: syn::DeriveInput) -> darling::Result { let opts = RepositoryOptions::from_derive_input(&ast)?; opts.columns.validate_list_for_by_columns()?; + opts.columns.validate_scope()?; opts.validate_forgettable()?; let repo = EsRepo::from(&opts); Ok(quote!(#repo)) @@ -206,6 +208,9 @@ impl ToTokens for EsRepo<'_> { let error_types = self.error_types.generate(); let map_constraint_fn = self.error_types.generate_map_constraint_fn(); + let scope_type = scope::ScopeType::new(self.opts); + let scope_type = quote! { #scope_type }; + let (impl_generics, ty_generics, where_clause) = self.generics.split_for_impl(); // If the event type has Forgettable fields, the repo must enable @@ -257,6 +262,8 @@ impl ToTokens for EsRepo<'_> { #error_types + #scope_type + #list_for_filters_struct #sort_by @@ -378,4 +385,157 @@ mod tests { }; assert!(derive(input).is_ok()); } + + #[test] + fn scoped_repo_is_ok() { + let input: syn::DeriveInput = parse_quote! { + #[es_repo( + entity = "User", + columns(partner_id(ty = "PartnerId", scope), name(ty = "String")) + )] + struct Users { + pool: sqlx::PgPool, + } + }; + let tokens = derive(input) + .expect("scoped repo should derive") + .to_string(); + assert!(tokens.contains("pub enum UserScope")); + // every read fn takes the scope argument + assert!(tokens.contains("fn find_by_id (& self , scope : impl Into < UserScope >")); + assert!(tokens.contains("(& self , scope : impl Into < UserScope > , ids")); + assert!(tokens.contains("fn list_by_created_at (& self , scope : impl Into < UserScope >")); + // the Only arm filters by the scope column, the All arm does not + assert!(tokens.contains("WHERE id = $1 AND partner_id = $2")); + assert!(tokens.contains("WHERE id = $1\"")); + // no find_by fns are generated for the scope column itself + assert!(!tokens.contains("find_by_partner_id")); + // writes stay unscoped (custody principle) + assert!(tokens.contains("fn create_in_op < OP > (& self , op : & mut OP , new_entity")); + assert!(tokens.contains("fn update_in_op < OP > (& self , op : & mut OP , entity")); + } + + #[test] + fn two_scope_columns_is_error() { + let input: syn::DeriveInput = parse_quote! { + #[es_repo( + entity = "User", + columns( + partner_id(ty = "PartnerId", scope), + customer_id(ty = "CustomerId", scope) + ) + )] + struct Users { + pool: sqlx::PgPool, + } + }; + let err = derive(input).unwrap_err(); + assert!( + err.to_string().contains("only one scope column"), + "unexpected error: {err}" + ); + } + + #[test] + fn optional_scope_column_is_error() { + let input: syn::DeriveInput = parse_quote! { + #[es_repo( + entity = "User", + columns(partner_id(ty = "Option", scope)) + )] + struct Users { + pool: sqlx::PgPool, + } + }; + let err = derive(input).unwrap_err(); + assert!( + err.to_string().contains("non-nullable"), + "unexpected error: {err}" + ); + } + + #[test] + fn nullable_annotated_scope_column_is_error() { + let input: syn::DeriveInput = parse_quote! { + #[es_repo( + entity = "User", + columns(partner_id(ty = "PartnerId", scope, nullable = true)) + )] + struct Users { + pool: sqlx::PgPool, + } + }; + let err = derive(input).unwrap_err(); + assert!( + err.to_string().contains("non-nullable"), + "unexpected error: {err}" + ); + } + + #[test] + fn scope_column_with_query_flags_is_error() { + for extra in ["find_by = true", "list_by = true", "list_for"] { + let src = format!( + r#" + #[es_repo( + entity = "User", + columns(partner_id(ty = "PartnerId", scope, {extra})) + )] + struct Users {{ + pool: sqlx::PgPool, + }} + "# + ); + let input: syn::DeriveInput = syn::parse_str(&src).unwrap(); + let err = derive(input).unwrap_err(); + assert!( + err.to_string() + .contains("cannot also be find_by, list_by or list_for"), + "unexpected error for `{extra}`: {err}" + ); + } + } + + #[test] + fn scope_on_nested_child_repo_is_error() { + let input: syn::DeriveInput = parse_quote! { + #[es_repo( + entity = "LineItem", + columns( + order_id(ty = "OrderId", parent), + partner_id(ty = "PartnerId", scope) + ) + )] + struct LineItems { + pool: sqlx::PgPool, + } + }; + let err = derive(input).unwrap_err(); + assert!( + err.to_string().contains("not supported on nested repos"), + "unexpected error: {err}" + ); + } + + #[test] + fn forgettable_scope_column_is_error() { + let input: syn::DeriveInput = parse_quote! { + #[es_repo( + entity = "User", + forgettable, + columns(partner_id(ty = "Forgettable", scope)) + )] + struct Users { + pool: sqlx::PgPool, + } + }; + let err = derive(input).unwrap_err(); + // Forgettable columns are rewritten to Option, so either the + // forgettable or the nullable check may fire first — both reject. + let msg = err.to_string(); + assert!( + msg.contains("Forgettable") || msg.contains("non-nullable"), + "unexpected error: {msg}" + ); + } } diff --git a/es-entity-macros/src/repo/options/columns.rs b/es-entity-macros/src/repo/options/columns.rs index cca5a98..af9328f 100644 --- a/es-entity-macros/src/repo/options/columns.rs +++ b/es-entity-macros/src/repo/options/columns.rs @@ -25,7 +25,9 @@ impl Columns { } pub fn all_find_by(&self) -> impl Iterator { - self.all.iter().filter(|c| c.opts.find_by()) + self.all + .iter() + .filter(|c| c.opts.find_by() && !c.opts.scope) } pub fn all_list_by(&self) -> impl Iterator { @@ -36,6 +38,69 @@ impl Columns { self.all.iter().filter(|c| c.opts.list_for()) } + /// The column marked `scope`, if any. Validated by + /// [`Self::validate_scope`] to be unique and non-nullable. + pub fn scope_column(&self) -> Option<&Column> { + self.all.iter().find(|c| c.opts.scope) + } + + /// Validates the `scope` column marker: + /// + /// - at most one column may be marked `scope` + /// - the scope column must be non-nullable (`Option` and + /// `nullable`-annotated types are rejected — nullable scope columns are + /// a future feature) + /// - the scope column must not be `Forgettable` + /// - the scope column must not also be a query column (`find_by`, + /// `list_by`, `list_for`) — every generated read is already filtered by + /// it, so per-tenant queries are the ordinary scoped fns + /// - the scope column must not be the `parent` column (nested repos + /// cannot be scoped — children are custody-guarded via their parent) + pub fn validate_scope(&self) -> darling::Result<()> { + let scope_columns: Vec<_> = self.all.iter().filter(|c| c.opts.scope).collect(); + if scope_columns.len() > 1 { + return Err(darling::Error::custom( + "only one scope column per repo is supported", + )); + } + let Some(col) = scope_columns.first() else { + return Ok(()); + }; + if col.is_nullable_column() { + return Err(darling::Error::custom(format!( + "scope column '{}' must be non-nullable — nullable scope columns are not supported (yet)", + col.name(), + ))); + } + if col.opts.forgettable { + return Err(darling::Error::custom(format!( + "scope column '{}' cannot be Forgettable", + col.name(), + ))); + } + if col.opts.find_by == Some(true) + || col.opts.list_by == Some(true) + || col.opts.list_for_opts.is_some() + { + return Err(darling::Error::custom(format!( + "scope column '{}' cannot also be find_by, list_by or list_for — every read is already filtered by it", + col.name(), + ))); + } + if col.opts.parent_opts.is_some() { + return Err(darling::Error::custom(format!( + "scope column '{}' cannot be the parent column — nested repos cannot be scoped", + col.name(), + ))); + } + if self.parent().is_some() { + return Err(darling::Error::custom( + "scope is not supported on nested repos — children are custody-guarded via their (scoped) parent", + )); + } + Ok(()) + } + pub fn find_list_by(&self, name: &syn::Ident) -> Option<&Column> { self.all .iter() @@ -459,6 +524,7 @@ impl Column { ty, is_id: true, forgettable: false, + scope: false, list_by: Some(true), find_by: Some(true), nullable: None, @@ -486,6 +552,7 @@ impl Column { ), is_id: false, forgettable: false, + scope: false, list_by: Some(true), find_by: Some(false), nullable: None, @@ -708,6 +775,11 @@ struct ColumnOpts { /// `NULL` by `forget()`/`delete()`. `ty` is rewritten to `Option`. #[darling(default, skip)] forgettable: bool, + /// Marks the repo's scope column: every generated read fn gains a leading + /// `scope: impl Into<{Entity}Scope>` argument and filters by this column + /// under `Only(_)`. Validated by [`Columns::validate_scope`]. + #[darling(default)] + scope: bool, #[darling(default)] find_by: Option, #[darling(default)] @@ -740,6 +812,7 @@ impl ColumnOpts { ty, is_id: false, forgettable: false, + scope: false, find_by: None, list_by: None, nullable: None, diff --git a/es-entity-macros/src/repo/options/mod.rs b/es-entity-macros/src/repo/options/mod.rs index 420e0ea..06467d7 100644 --- a/es-entity-macros/src/repo/options/mod.rs +++ b/es-entity-macros/src/repo/options/mod.rs @@ -465,6 +465,13 @@ impl RepositoryOptions { syn::Ident::new(&format!("{}Column", self.entity_ident), Span::call_site()) } + /// The generated scope enum ident (`{Entity}Scope`), entity-named like + /// the other generated companion types (`{Entity}FindError`, + /// `{Entity}ByIdCursor`, ...). + pub fn scope_type_ident(&self) -> syn::Ident { + syn::Ident::new(&format!("{}Scope", self.entity_ident), Span::call_site()) + } + pub fn query_fn_get_op(nested: bool) -> proc_macro2::TokenStream { if nested { quote! { diff --git a/es-entity-macros/src/repo/scope.rs b/es-entity-macros/src/repo/scope.rs new file mode 100644 index 0000000..a8fcbf3 --- /dev/null +++ b/es-entity-macros/src/repo/scope.rs @@ -0,0 +1,190 @@ +use darling::ToTokens; +use proc_macro2::TokenStream; +use quote::{TokenStreamExt, quote}; + +use super::options::*; + +/// Info about a repo's scope column, shared by the read-fn emitters. +/// +/// A repo with a column marked `scope` generates every read fn +/// (`find_by_*`, `find_all`, `list_by_*`, `list_for_*`, `list_for_filters*`) +/// with a leading `scope: impl Into<{Entity}Scope>` argument. At runtime the +/// fn dispatches on the scope: `All` executes the exact same SQL as an +/// unscoped repo, `Only(value)` executes a variant with an additional +/// `scope_column = $n` conjunct — both static, sargable `es_query!` literals. +#[derive(Clone)] +pub struct ScopeInfo<'a> { + /// The generated scope enum ident: `{Entity}Scope`. + pub scope_ty: syn::Ident, + pub column_name: &'a syn::Ident, + pub column_ty: &'a syn::Type, +} + +impl<'a> ScopeInfo<'a> { + pub fn from_opts(opts: &'a RepositoryOptions) -> Option { + opts.columns.scope_column().map(|col| ScopeInfo { + scope_ty: opts.scope_type_ident(), + column_name: col.name(), + column_ty: col.ty(), + }) + } + + /// The `scope: impl Into<{Entity}Scope>,` fn argument. + pub fn fn_arg(&self) -> TokenStream { + let scope_ty = &self.scope_ty; + quote! { scope: impl Into<#scope_ty>, } + } + + /// Forwarding token for the standalone -> `_in_op` delegation. + pub fn fn_pass(&self) -> TokenStream { + quote! { scope, } + } + + /// Converts the `impl Into<_>` argument once at fn entry. + pub fn convert(&self) -> TokenStream { + quote! { let __scope = scope.into(); } + } + + /// The SQL conjunct for the `Only` arm at the given parameter index. + pub fn predicate(&self, param_idx: u32) -> String { + format!("{} = ${}", self.column_name, param_idx) + } + + /// The query binding for the `Only` arm (pairs with [`Self::dispatch`]'s + /// `__scope_val` pattern binding). + pub fn arg_tokens(&self) -> TokenStream { + let column_ty = self.column_ty; + quote! { __scope_val as &#column_ty, } + } + + /// Runtime dispatch between the unscoped (`All`) and scoped (`Only`) + /// query variants. The `Only` arm binds `__scope_val: &T`. + pub fn dispatch(&self, all_arm: TokenStream, only_arm: TokenStream) -> TokenStream { + let scope_ty = &self.scope_ty; + quote! { + match &__scope { + #scope_ty::All => #all_arm, + #scope_ty::Only(__scope_val) => #only_arm, + } + } + } +} + +/// Generates the per-repo scope enum: +/// +/// ```ignore +/// pub enum UserScope { +/// All, +/// Only(PartnerId), +/// } +/// ``` +/// +/// plus `From` / `From<&T>` conversions into `Only` so call sites can pass +/// a scope value directly. Deliberately **no** `From>`: mapping +/// `None` to `All` would turn a stray `None` into silent all-scope access. +pub struct ScopeType<'a> { + entity: &'a syn::Ident, + info: ScopeInfo<'a>, +} + +impl<'a> ScopeType<'a> { + pub fn new(opts: &'a RepositoryOptions) -> Option { + ScopeInfo::from_opts(opts).map(|info| Self { + entity: opts.entity(), + info, + }) + } +} + +impl ToTokens for ScopeType<'_> { + fn to_tokens(&self, tokens: &mut TokenStream) { + let scope_ty = &self.info.scope_ty; + let column_ty = self.info.column_ty; + let doc = format!( + "Scope argument for [`{}`] repository reads: `Only(value)` filters every query by \ + the scope column, `All` reads across all scopes (audited escape hatch).", + self.entity, + ); + + tokens.append_all(quote! { + #[doc = #doc] + #[derive(Debug, Clone, Copy)] + pub enum #scope_ty { + /// No scope filter — reads across all scopes. + All, + /// Restricts every read to rows whose scope column equals the value. + Only(#column_ty), + } + + impl From<#column_ty> for #scope_ty { + fn from(value: #column_ty) -> Self { + #scope_ty::Only(value) + } + } + + impl From<&#column_ty> for #scope_ty { + fn from(value: &#column_ty) -> Self { + #scope_ty::Only(*value) + } + } + + impl From<&#scope_ty> for #scope_ty { + fn from(value: &#scope_ty) -> Self { + *value + } + } + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use proc_macro2::Span; + + fn test_info<'a>( + column_name: &'a syn::Ident, + column_ty: &'a syn::Type, + ) -> (ScopeInfo<'a>, syn::Ident) { + let entity = syn::Ident::new("Entity", Span::call_site()); + ( + ScopeInfo { + scope_ty: syn::Ident::new("EntityScope", Span::call_site()), + column_name, + column_ty, + }, + entity, + ) + } + + #[test] + fn scope_type_tokens() { + let column_name = syn::Ident::new("partner_id", Span::call_site()); + let column_ty: syn::Type = syn::parse_str("PartnerId").unwrap(); + let (info, entity) = test_info(&column_name, &column_ty); + let scope_type = ScopeType { + entity: &entity, + info, + }; + + let mut tokens = TokenStream::new(); + scope_type.to_tokens(&mut tokens); + let token_str = tokens.to_string(); + + assert!(token_str.contains("pub enum EntityScope")); + assert!(token_str.contains("Only (PartnerId)")); + assert!(token_str.contains("impl From < PartnerId > for EntityScope")); + assert!(token_str.contains("impl From < & PartnerId > for EntityScope")); + assert!(!token_str.contains("Option")); + } + + #[test] + fn scope_info_predicate() { + let column_name = syn::Ident::new("partner_id", Span::call_site()); + let column_ty: syn::Type = syn::parse_str("PartnerId").unwrap(); + let (info, _entity) = test_info(&column_name, &column_ty); + + assert_eq!(info.predicate(1), "partner_id = $1"); + assert_eq!(info.predicate(3), "partner_id = $3"); + } +} diff --git a/migrations/20260730000000_scoped_repo_test.sql b/migrations/20260730000000_scoped_repo_test.sql new file mode 100644 index 0000000..74d5bc7 --- /dev/null +++ b/migrations/20260730000000_scoped_repo_test.sql @@ -0,0 +1,19 @@ +CREATE TABLE contacts ( + id UUID PRIMARY KEY, + partner_id UUID NOT NULL, + email VARCHAR NOT NULL, + status VARCHAR NOT NULL, + created_at TIMESTAMPTZ NOT NULL +); +CREATE INDEX idx_contacts_partner_created_id ON contacts (partner_id, created_at DESC, id DESC); +CREATE INDEX idx_contacts_partner_email ON contacts (partner_id, email); + +CREATE TABLE contact_events ( + id UUID NOT NULL REFERENCES contacts(id), + sequence INT NOT NULL, + event_type VARCHAR NOT NULL, + event JSONB NOT NULL, + context JSONB DEFAULT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + UNIQUE(id, sequence) +); diff --git a/tests/entities/contact.rs b/tests/entities/contact.rs new file mode 100644 index 0000000..2795a04 --- /dev/null +++ b/tests/entities/contact.rs @@ -0,0 +1,90 @@ +#![allow(dead_code)] + +use derive_builder::Builder; +use serde::{Deserialize, Serialize}; + +use es_entity::*; + +es_entity::entity_id! { ContactId } +es_entity::entity_id! { PartnerId } + +/// Mirrors the shape of lana's partner-scoped entities: a non-nullable +/// tenant column (`partner_id`) marked `scope`, a unique lookup column +/// (`email`) and a filter column (`status`). +#[derive(EsEvent, Debug, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +#[es_event(id = "ContactId")] +pub enum ContactEvent { + Initialized { + id: ContactId, + partner_id: PartnerId, + email: String, + status: String, + }, +} + +#[derive(EsEntity, Builder)] +#[builder(pattern = "owned", build_fn(error = "EntityHydrationError"))] +pub struct Contact { + pub id: ContactId, + pub partner_id: PartnerId, + pub email: String, + pub status: String, + + events: EntityEvents, +} + +impl TryFromEvents for Contact { + fn try_from_events(events: EntityEvents) -> Result { + let mut builder = ContactBuilder::default(); + for event in events.iter_all() { + match event { + ContactEvent::Initialized { + id, + partner_id, + email, + status, + } => { + builder = builder + .id(*id) + .partner_id(*partner_id) + .email(email.clone()) + .status(status.clone()); + } + } + } + builder.events(events).build() + } +} + +#[derive(Debug, Builder)] +pub struct NewContact { + #[builder(setter(into))] + pub id: ContactId, + #[builder(setter(into))] + pub partner_id: PartnerId, + #[builder(setter(into))] + pub email: String, + #[builder(setter(into))] + pub status: String, +} + +impl NewContact { + pub fn builder() -> NewContactBuilder { + NewContactBuilder::default() + } +} + +impl IntoEvents for NewContact { + fn into_events(self) -> EntityEvents { + EntityEvents::init( + self.id, + [ContactEvent::Initialized { + id: self.id, + partner_id: self.partner_id, + email: self.email, + status: self.status, + }], + ) + } +} diff --git a/tests/entities/mod.rs b/tests/entities/mod.rs index 3623ab0..ceac83f 100644 --- a/tests/entities/mod.rs +++ b/tests/entities/mod.rs @@ -1,3 +1,4 @@ +pub mod contact; pub mod customer; pub mod order; pub mod profile; diff --git a/tests/scoped_repo.rs b/tests/scoped_repo.rs new file mode 100644 index 0000000..5104bc6 --- /dev/null +++ b/tests/scoped_repo.rs @@ -0,0 +1,372 @@ +mod entities; +mod helpers; + +use sqlx::PgPool; + +use entities::contact::*; +use es_entity::*; + +/// Partner-scoped repo: every generated read fn requires a leading +/// `scope: impl Into` argument. `Only(partner_id)` filters all +/// reads by the scope column; `All` reads across scopes. Writes (`create`, +/// `update`, `delete`) keep their unscoped signatures — mutations operate on +/// entities that could only have been obtained through a scoped read. +#[derive(EsRepo, Debug)] +#[es_repo( + entity = "Contact", + columns( + partner_id(ty = "PartnerId", scope), + email(ty = "String"), + status(ty = "String", list_for(by(created_at))), + ) +)] +pub struct Contacts { + pool: PgPool, +} + +impl Contacts { + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +async fn seed_contacts( + repo: &Contacts, + partner_id: PartnerId, + specs: &[(&str, &str)], +) -> anyhow::Result> { + let mut ids = Vec::new(); + for (email, status) in specs { + let id = ContactId::new(); + let new = NewContact::builder() + .id(id) + .partner_id(partner_id) + .email(format!("{email}-{id}@test.com")) + .status(*status) + .build() + .unwrap(); + // create is deliberately unscoped: the scope column is ordinary + // NewEntity data. + repo.create(new).await?; + ids.push(id); + } + Ok(ids) +} + +#[tokio::test] +async fn scoped_point_reads() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let contacts = Contacts::new(pool); + + let partner_a = PartnerId::new(); + let partner_b = PartnerId::new(); + let ids_a = seed_contacts(&contacts, partner_a, &[("a", "active")]).await?; + let ids_b = seed_contacts(&contacts, partner_b, &[("b", "active")]).await?; + let (id_a, id_b) = (ids_a[0], ids_b[0]); + + // own scope: found — `impl Into` accepts the partner id + // directly (From => Only), a reference, or the explicit enum. + let found = contacts.find_by_id(partner_a, id_a).await?; + assert_eq!(found.partner_id, partner_a); + contacts.find_by_id(&partner_a, id_a).await?; + contacts + .find_by_id(ContactScope::Only(partner_a), id_a) + .await?; + + // foreign scope: missing and not-yours look identical + let err = contacts.find_by_id(partner_b, id_a).await; + assert!(matches!(err, Err(ContactFindError::NotFound { .. }))); + assert!(contacts.maybe_find_by_id(partner_b, id_a).await?.is_none()); + + // All: reads across scopes (audited escape hatch) + contacts.find_by_id(ContactScope::All, id_a).await?; + contacts.find_by_id(ContactScope::All, id_b).await?; + + // unique-column lookup is scoped the same way + let email = contacts.find_by_id(partner_a, id_a).await?.email; + contacts.find_by_email(partner_a, &email).await?; + assert!( + contacts + .maybe_find_by_email(partner_b, &email) + .await? + .is_none() + ); + + Ok(()) +} + +#[tokio::test] +async fn scoped_reads_in_op() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let contacts = Contacts::new(pool); + + let partner_a = PartnerId::new(); + let partner_b = PartnerId::new(); + let ids = seed_contacts(&contacts, partner_a, &[("op", "active")]).await?; + + let mut op = contacts.begin_op().await?; + contacts + .find_by_id_in_op(&mut op, partner_a, ids[0]) + .await?; + assert!( + contacts + .maybe_find_by_id_in_op(&mut op, partner_b, ids[0]) + .await? + .is_none() + ); + op.commit().await?; + + Ok(()) +} + +#[tokio::test] +async fn scoped_find_all_drops_foreign_ids() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let contacts = Contacts::new(pool); + + let partner_a = PartnerId::new(); + let partner_b = PartnerId::new(); + let ids_a = seed_contacts( + &contacts, + partner_a, + &[("fa1", "active"), ("fa2", "active")], + ) + .await?; + let ids_b = seed_contacts(&contacts, partner_b, &[("fb1", "active")]).await?; + + let all_ids: Vec = ids_a.iter().chain(ids_b.iter()).copied().collect(); + + // Only(a): foreign ids are silently absent — missing and not-yours look + // identical. + let found = contacts.find_all::(partner_a, &all_ids).await?; + assert_eq!(found.len(), 2); + assert!(ids_a.iter().all(|id| found.contains_key(id))); + assert!(ids_b.iter().all(|id| !found.contains_key(id))); + + // All: everything. + let found = contacts + .find_all::(ContactScope::All, &all_ids) + .await?; + assert_eq!(found.len(), 3); + + Ok(()) +} + +#[tokio::test] +async fn scoped_list_by_paginates_within_scope() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let contacts = Contacts::new(pool); + + let partner_a = PartnerId::new(); + let partner_b = PartnerId::new(); + let specs: Vec<(&str, &str)> = (0..5).map(|_| ("list", "active")).collect(); + let ids_a = seed_contacts(&contacts, partner_a, &specs).await?; + seed_contacts(&contacts, partner_b, &specs).await?; + + // paginate under Only(a) with a page size that forces both the page-1 + // and the cursor-page specialized variants to execute + let mut collected = Vec::new(); + let mut after: Option = None; + let mut pages = 0; + loop { + let ret = contacts + .list_by_created_at( + partner_a, + PaginatedQueryArgs { first: 2, after }, + ListDirection::Descending, + ) + .await?; + pages += 1; + for entity in &ret.entities { + assert_eq!( + entity.partner_id, partner_a, + "scoped list leaked a foreign row" + ); + } + collected.extend(ret.entities.iter().map(|c| c.id)); + if !ret.has_next_page { + break; + } + after = ret.end_cursor; + } + assert!(pages >= 3, "expected pagination across pages, got {pages}"); + assert_eq!(collected.len(), 5); + let expected: std::collections::HashSet<_> = ids_a.into_iter().collect(); + let collected: std::collections::HashSet<_> = collected.into_iter().collect(); + assert_eq!(collected, expected); + + Ok(()) +} + +#[tokio::test] +async fn scoped_list_for_and_filters_dispatch() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let contacts = Contacts::new(pool); + + let partner_a = PartnerId::new(); + let partner_b = PartnerId::new(); + seed_contacts( + &contacts, + partner_a, + &[("d1", "active"), ("d2", "inactive"), ("d3", "active")], + ) + .await?; + seed_contacts(&contacts, partner_b, &[("d4", "active")]).await?; + + // dedicated single-filter path + let ret = contacts + .list_for_status_by_created_at( + partner_a, + "active", + PaginatedQueryArgs { + first: 100, + after: None, + }, + ListDirection::Descending, + ) + .await?; + assert!(ret.entities.iter().all(|c| c.partner_id == partner_a)); + assert!(ret.entities.iter().all(|c| c.status == "active")); + assert_eq!(ret.entities.len(), 2); + + // unified dispatch: no filter routes through the (scoped) list_by proxy + let ret = contacts + .list_for_filters( + partner_a, + ContactFilters::default(), + Sort { + by: ContactSortBy::CreatedAt, + direction: ListDirection::Descending, + }, + PaginatedQueryArgs { + first: 100, + after: None, + }, + ) + .await?; + assert_eq!(ret.entities.len(), 3); + assert!(ret.entities.iter().all(|c| c.partner_id == partner_a)); + + // unified dispatch: status filter routes through the (scoped) list_for proxy + let ret = contacts + .list_for_filters( + partner_a, + ContactFilters { + status: Some("inactive".to_string()), + }, + Sort { + by: ContactSortBy::CreatedAt, + direction: ListDirection::Descending, + }, + PaginatedQueryArgs { + first: 100, + after: None, + }, + ) + .await?; + assert_eq!(ret.entities.len(), 1); + assert_eq!(ret.entities[0].partner_id, partner_a); + + // All sees both partners' rows (restricted to this test's seeds via status) + let ret = contacts + .list_for_status_by_created_at( + ContactScope::All, + "inactive", + PaginatedQueryArgs { + first: 100, + after: None, + }, + ListDirection::Descending, + ) + .await?; + assert!(ret.entities.iter().any(|c| c.partner_id == partner_a)); + + Ok(()) +} + +/// A cursor minted under one scope replayed under another repositions the +/// pagination but can never leak foreign rows — every page's SQL carries the +/// scope conjunct. +#[tokio::test] +async fn foreign_cursor_cannot_leak_rows() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + let contacts = Contacts::new(pool); + + let partner_a = PartnerId::new(); + let partner_b = PartnerId::new(); + let specs: Vec<(&str, &str)> = (0..3).map(|_| ("cur", "active")).collect(); + seed_contacts(&contacts, partner_a, &specs).await?; + seed_contacts(&contacts, partner_b, &specs).await?; + + let page_a = contacts + .list_by_created_at( + partner_a, + PaginatedQueryArgs { + first: 1, + after: None, + }, + ListDirection::Descending, + ) + .await?; + let cursor_from_a = page_a.end_cursor; + + let ret = contacts + .list_by_created_at( + partner_b, + PaginatedQueryArgs { + first: 100, + after: cursor_from_a, + }, + ListDirection::Descending, + ) + .await?; + assert!( + ret.entities.iter().all(|c| c.partner_id == partner_b), + "foreign cursor must not leak rows from another scope" + ); + + Ok(()) +} + +/// The `Only` arm's page-1 list SQL must be sargable against a +/// `(partner_id, created_at, id)` composite index — the scope conjunct is a +/// plain equality, not a COALESCE catch-all. +#[tokio::test] +async fn scoped_list_query_plan_uses_composite_index() -> anyhow::Result<()> { + let pool = helpers::init_pool().await?; + + let mut tx = pool.begin().await?; + sqlx::query("SET LOCAL enable_seqscan = off") + .execute(&mut *tx) + .await?; + // The exact page-1 `Only` arm SQL shape generated for + // `list_by_created_at`. + let rows: Vec<(String,)> = sqlx::query_as( + "EXPLAIN SELECT created_at, id FROM contacts WHERE partner_id = $1 \ + ORDER BY created_at DESC, id DESC LIMIT $2", + ) + .bind(uuid::Uuid::from(PartnerId::new())) + .bind(5i64) + .fetch_all(&mut *tx) + .await?; + tx.rollback().await?; + + let plan = rows + .into_iter() + .map(|(l,)| l) + .collect::>() + .join("\n"); + // The exact index the planner picks depends on table statistics; the + // property under test is sargability — the scope conjunct must become an + // index qual, never a filter over a seq scan. + assert!( + plan.contains("Index Cond: (partner_id ="), + "expected an index qual on partner_id, got plan:\n{plan}" + ); + assert!( + !plan.contains("Seq Scan"), + "scope conjunct must not force a sequential scan, got plan:\n{plan}" + ); + + Ok(()) +}