better_auth_core/adapters/
database.rs

1use async_trait::async_trait;
2use chrono::{DateTime, Utc};
3use std::collections::HashMap;
4use std::sync::{Arc, Mutex};
5use uuid::Uuid;
6
7use crate::entity::{
8    AuthAccount, AuthInvitation, AuthMember, AuthOrganization, AuthSession, AuthUser,
9    AuthVerification,
10};
11use crate::error::{AuthError, AuthResult};
12use crate::types::{
13    Account, CreateAccount, CreateInvitation, CreateMember, CreateOrganization, CreateSession,
14    CreateUser, CreateVerification, Invitation, InvitationStatus, Member, Organization, Session,
15    UpdateOrganization, UpdateUser, User, Verification,
16};
17
18/// Database adapter trait for persistence.
19///
20/// Associated types allow users to define their own entity structs.
21/// Use the default types (`User`, `Session`, etc.) or implement entity traits
22/// on custom structs via `#[derive(AuthUser)]` etc.
23#[async_trait]
24pub trait DatabaseAdapter: Send + Sync + 'static {
25    type User: AuthUser;
26    type Session: AuthSession;
27    type Account: AuthAccount;
28    type Organization: AuthOrganization;
29    type Member: AuthMember;
30    type Invitation: AuthInvitation;
31    type Verification: AuthVerification;
32
33    // User operations
34    async fn create_user(&self, user: CreateUser) -> AuthResult<Self::User>;
35    async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<Self::User>>;
36    async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<Self::User>>;
37    async fn get_user_by_username(&self, username: &str) -> AuthResult<Option<Self::User>>;
38    async fn update_user(&self, id: &str, update: UpdateUser) -> AuthResult<Self::User>;
39    async fn delete_user(&self, id: &str) -> AuthResult<()>;
40
41    // Session operations
42    async fn create_session(&self, session: CreateSession) -> AuthResult<Self::Session>;
43    async fn get_session(&self, token: &str) -> AuthResult<Option<Self::Session>>;
44    async fn get_user_sessions(&self, user_id: &str) -> AuthResult<Vec<Self::Session>>;
45    async fn update_session_expiry(&self, token: &str, expires_at: DateTime<Utc>)
46    -> AuthResult<()>;
47    async fn delete_session(&self, token: &str) -> AuthResult<()>;
48    async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()>;
49    async fn delete_expired_sessions(&self) -> AuthResult<usize>;
50
51    // Account operations (for OAuth)
52    async fn create_account(&self, account: CreateAccount) -> AuthResult<Self::Account>;
53    async fn get_account(
54        &self,
55        provider: &str,
56        provider_account_id: &str,
57    ) -> AuthResult<Option<Self::Account>>;
58    async fn get_user_accounts(&self, user_id: &str) -> AuthResult<Vec<Self::Account>>;
59    async fn delete_account(&self, id: &str) -> AuthResult<()>;
60
61    // Verification token operations
62    async fn create_verification(
63        &self,
64        verification: CreateVerification,
65    ) -> AuthResult<Self::Verification>;
66    async fn get_verification(
67        &self,
68        identifier: &str,
69        value: &str,
70    ) -> AuthResult<Option<Self::Verification>>;
71    async fn get_verification_by_value(
72        &self,
73        value: &str,
74    ) -> AuthResult<Option<Self::Verification>>;
75    async fn delete_verification(&self, id: &str) -> AuthResult<()>;
76    async fn delete_expired_verifications(&self) -> AuthResult<usize>;
77
78    // Organization operations
79    async fn create_organization(&self, org: CreateOrganization) -> AuthResult<Self::Organization>;
80    async fn get_organization_by_id(&self, id: &str) -> AuthResult<Option<Self::Organization>>;
81    async fn get_organization_by_slug(&self, slug: &str) -> AuthResult<Option<Self::Organization>>;
82    async fn update_organization(
83        &self,
84        id: &str,
85        update: UpdateOrganization,
86    ) -> AuthResult<Self::Organization>;
87    async fn delete_organization(&self, id: &str) -> AuthResult<()>;
88    async fn list_user_organizations(&self, user_id: &str) -> AuthResult<Vec<Self::Organization>>;
89
90    // Member operations
91    async fn create_member(&self, member: CreateMember) -> AuthResult<Self::Member>;
92    async fn get_member(
93        &self,
94        organization_id: &str,
95        user_id: &str,
96    ) -> AuthResult<Option<Self::Member>>;
97    async fn get_member_by_id(&self, id: &str) -> AuthResult<Option<Self::Member>>;
98    async fn update_member_role(&self, member_id: &str, role: &str) -> AuthResult<Self::Member>;
99    async fn delete_member(&self, member_id: &str) -> AuthResult<()>;
100    async fn list_organization_members(
101        &self,
102        organization_id: &str,
103    ) -> AuthResult<Vec<Self::Member>>;
104    async fn count_organization_members(&self, organization_id: &str) -> AuthResult<usize>;
105    async fn count_organization_owners(&self, organization_id: &str) -> AuthResult<usize>;
106
107    // Invitation operations
108    async fn create_invitation(&self, invitation: CreateInvitation)
109    -> AuthResult<Self::Invitation>;
110    async fn get_invitation_by_id(&self, id: &str) -> AuthResult<Option<Self::Invitation>>;
111    async fn get_pending_invitation(
112        &self,
113        organization_id: &str,
114        email: &str,
115    ) -> AuthResult<Option<Self::Invitation>>;
116    async fn update_invitation_status(
117        &self,
118        id: &str,
119        status: InvitationStatus,
120    ) -> AuthResult<Self::Invitation>;
121    async fn list_organization_invitations(
122        &self,
123        organization_id: &str,
124    ) -> AuthResult<Vec<Self::Invitation>>;
125    async fn list_user_invitations(&self, email: &str) -> AuthResult<Vec<Self::Invitation>>;
126
127    // Session organization support
128    async fn update_session_active_organization(
129        &self,
130        token: &str,
131        organization_id: Option<&str>,
132    ) -> AuthResult<Self::Session>;
133}
134
135/// In-memory database adapter for testing and development
136pub struct MemoryDatabaseAdapter {
137    users: Arc<Mutex<HashMap<String, User>>>,
138    sessions: Arc<Mutex<HashMap<String, Session>>>,
139    accounts: Arc<Mutex<HashMap<String, Account>>>,
140    verifications: Arc<Mutex<HashMap<String, Verification>>>,
141    email_index: Arc<Mutex<HashMap<String, String>>>, // email -> user_id
142    username_index: Arc<Mutex<HashMap<String, String>>>, // username -> user_id
143    // Organization data
144    organizations: Arc<Mutex<HashMap<String, Organization>>>,
145    members: Arc<Mutex<HashMap<String, Member>>>,
146    invitations: Arc<Mutex<HashMap<String, Invitation>>>,
147    slug_index: Arc<Mutex<HashMap<String, String>>>, // slug -> organization_id
148}
149
150impl MemoryDatabaseAdapter {
151    pub fn new() -> Self {
152        Self {
153            users: Arc::new(Mutex::new(HashMap::new())),
154            sessions: Arc::new(Mutex::new(HashMap::new())),
155            accounts: Arc::new(Mutex::new(HashMap::new())),
156            verifications: Arc::new(Mutex::new(HashMap::new())),
157            email_index: Arc::new(Mutex::new(HashMap::new())),
158            username_index: Arc::new(Mutex::new(HashMap::new())),
159            organizations: Arc::new(Mutex::new(HashMap::new())),
160            members: Arc::new(Mutex::new(HashMap::new())),
161            invitations: Arc::new(Mutex::new(HashMap::new())),
162            slug_index: Arc::new(Mutex::new(HashMap::new())),
163        }
164    }
165}
166
167impl Default for MemoryDatabaseAdapter {
168    fn default() -> Self {
169        Self::new()
170    }
171}
172
173#[async_trait]
174impl DatabaseAdapter for MemoryDatabaseAdapter {
175    type User = User;
176    type Session = Session;
177    type Account = Account;
178    type Organization = Organization;
179    type Member = Member;
180    type Invitation = Invitation;
181    type Verification = Verification;
182
183    async fn create_user(&self, create_user: CreateUser) -> AuthResult<User> {
184        let mut users = self.users.lock().unwrap();
185        let mut email_index = self.email_index.lock().unwrap();
186        let mut username_index = self.username_index.lock().unwrap();
187
188        let id = create_user.id.unwrap_or_else(|| Uuid::new_v4().to_string());
189
190        // Check if email already exists
191        if let Some(email) = &create_user.email
192            && email_index.contains_key(email)
193        {
194            return Err(AuthError::config("Email already exists"));
195        }
196
197        // Check if username already exists
198        if let Some(username) = &create_user.username
199            && username_index.contains_key(username)
200        {
201            return Err(AuthError::conflict(
202                "A user with this username already exists",
203            ));
204        }
205
206        let now = Utc::now();
207        let user = User {
208            id: id.clone(),
209            name: create_user.name,
210            email: create_user.email.clone(),
211            email_verified: create_user.email_verified.unwrap_or(false),
212            image: create_user.image,
213            created_at: now,
214            updated_at: now,
215            username: create_user.username.clone(),
216            display_username: create_user.display_username,
217            two_factor_enabled: false,
218            role: create_user.role,
219            banned: false,
220            ban_reason: None,
221            ban_expires: None,
222            metadata: create_user.metadata.unwrap_or_default(),
223        };
224
225        users.insert(id.clone(), user.clone());
226
227        if let Some(email) = &create_user.email {
228            email_index.insert(email.clone(), id.clone());
229        }
230
231        if let Some(username) = &create_user.username {
232            username_index.insert(username.clone(), id);
233        }
234
235        Ok(user)
236    }
237
238    async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<User>> {
239        let users = self.users.lock().unwrap();
240        Ok(users.get(id).cloned())
241    }
242
243    async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<User>> {
244        let email_index = self.email_index.lock().unwrap();
245        let users = self.users.lock().unwrap();
246
247        if let Some(user_id) = email_index.get(email) {
248            Ok(users.get(user_id).cloned())
249        } else {
250            Ok(None)
251        }
252    }
253
254    async fn get_user_by_username(&self, username: &str) -> AuthResult<Option<User>> {
255        let username_index = self.username_index.lock().unwrap();
256        let users = self.users.lock().unwrap();
257
258        if let Some(user_id) = username_index.get(username) {
259            Ok(users.get(user_id).cloned())
260        } else {
261            Ok(None)
262        }
263    }
264
265    async fn update_user(&self, id: &str, update: UpdateUser) -> AuthResult<User> {
266        let mut users = self.users.lock().unwrap();
267        let mut email_index = self.email_index.lock().unwrap();
268        let mut username_index = self.username_index.lock().unwrap();
269
270        let user = users.get_mut(id).ok_or(AuthError::UserNotFound)?;
271
272        // Update email index if email changed
273        if let Some(new_email) = &update.email {
274            if let Some(old_email) = &user.email {
275                email_index.remove(old_email);
276            }
277            email_index.insert(new_email.clone(), id.to_string());
278            user.email = Some(new_email.clone());
279        }
280
281        if let Some(name) = update.name {
282            user.name = Some(name);
283        }
284
285        if let Some(image) = update.image {
286            user.image = Some(image);
287        }
288
289        if let Some(email_verified) = update.email_verified {
290            user.email_verified = email_verified;
291        }
292
293        if let Some(ref username) = update.username {
294            // Update username index
295            if let Some(old_username) = &user.username {
296                username_index.remove(old_username);
297            }
298            username_index.insert(username.clone(), id.to_string());
299            user.username = Some(username.clone());
300        }
301
302        if let Some(display_username) = update.display_username {
303            user.display_username = Some(display_username);
304        }
305
306        if let Some(role) = update.role {
307            user.role = Some(role);
308        }
309
310        if let Some(banned) = update.banned {
311            user.banned = banned;
312        }
313
314        if let Some(ban_reason) = update.ban_reason {
315            user.ban_reason = Some(ban_reason);
316        }
317
318        if let Some(ban_expires) = update.ban_expires {
319            user.ban_expires = Some(ban_expires);
320        }
321
322        if let Some(two_factor_enabled) = update.two_factor_enabled {
323            user.two_factor_enabled = two_factor_enabled;
324        }
325
326        if let Some(metadata) = update.metadata {
327            user.metadata = metadata;
328        }
329
330        user.updated_at = Utc::now();
331
332        Ok(user.clone())
333    }
334
335    async fn delete_user(&self, id: &str) -> AuthResult<()> {
336        let mut users = self.users.lock().unwrap();
337        let mut email_index = self.email_index.lock().unwrap();
338        let mut username_index = self.username_index.lock().unwrap();
339
340        if let Some(user) = users.remove(id) {
341            if let Some(email) = &user.email {
342                email_index.remove(email);
343            }
344            if let Some(username) = &user.username {
345                username_index.remove(username);
346            }
347        }
348
349        Ok(())
350    }
351
352    async fn create_session(&self, create_session: CreateSession) -> AuthResult<Session> {
353        let mut sessions = self.sessions.lock().unwrap();
354
355        let token = format!("session_{}", Uuid::new_v4());
356        let now = Utc::now();
357        let session = Session {
358            id: Uuid::new_v4().to_string(),
359            expires_at: create_session.expires_at,
360            token: token.clone(),
361            created_at: now,
362            updated_at: now,
363            ip_address: create_session.ip_address,
364            user_agent: create_session.user_agent,
365            user_id: create_session.user_id,
366            impersonated_by: create_session.impersonated_by,
367            active_organization_id: create_session.active_organization_id,
368            active: true,
369        };
370
371        sessions.insert(token, session.clone());
372        Ok(session)
373    }
374
375    async fn get_session(&self, token: &str) -> AuthResult<Option<Session>> {
376        let sessions = self.sessions.lock().unwrap();
377        Ok(sessions.get(token).cloned())
378    }
379
380    async fn get_user_sessions(&self, user_id: &str) -> AuthResult<Vec<Session>> {
381        let sessions = self.sessions.lock().unwrap();
382        Ok(sessions
383            .values()
384            .filter(|session| session.user_id == user_id && session.active)
385            .cloned()
386            .collect())
387    }
388
389    async fn update_session_expiry(
390        &self,
391        token: &str,
392        expires_at: DateTime<Utc>,
393    ) -> AuthResult<()> {
394        let mut sessions = self.sessions.lock().unwrap();
395        if let Some(session) = sessions.get_mut(token) {
396            session.expires_at = expires_at;
397        }
398        Ok(())
399    }
400
401    async fn delete_session(&self, token: &str) -> AuthResult<()> {
402        let mut sessions = self.sessions.lock().unwrap();
403        sessions.remove(token);
404        Ok(())
405    }
406
407    async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()> {
408        let mut sessions = self.sessions.lock().unwrap();
409        sessions.retain(|_, session| session.user_id != user_id);
410        Ok(())
411    }
412
413    async fn delete_expired_sessions(&self) -> AuthResult<usize> {
414        let mut sessions = self.sessions.lock().unwrap();
415        let now = Utc::now();
416        let initial_count = sessions.len();
417
418        sessions.retain(|_, session| session.expires_at > now && session.active);
419
420        Ok(initial_count - sessions.len())
421    }
422
423    async fn create_account(&self, create_account: CreateAccount) -> AuthResult<Account> {
424        let mut accounts = self.accounts.lock().unwrap();
425
426        let now = Utc::now();
427        let account = Account {
428            id: Uuid::new_v4().to_string(),
429            account_id: create_account.account_id,
430            provider_id: create_account.provider_id,
431            user_id: create_account.user_id,
432            access_token: create_account.access_token,
433            refresh_token: create_account.refresh_token,
434            id_token: create_account.id_token,
435            access_token_expires_at: create_account.access_token_expires_at,
436            refresh_token_expires_at: create_account.refresh_token_expires_at,
437            scope: create_account.scope,
438            password: create_account.password,
439            created_at: now,
440            updated_at: now,
441        };
442
443        accounts.insert(account.id.clone(), account.clone());
444        Ok(account)
445    }
446
447    async fn get_account(
448        &self,
449        provider: &str,
450        provider_account_id: &str,
451    ) -> AuthResult<Option<Account>> {
452        let accounts = self.accounts.lock().unwrap();
453        Ok(accounts
454            .values()
455            .find(|acc| acc.provider_id == provider && acc.account_id == provider_account_id)
456            .cloned())
457    }
458
459    async fn get_user_accounts(&self, user_id: &str) -> AuthResult<Vec<Account>> {
460        let accounts = self.accounts.lock().unwrap();
461        Ok(accounts
462            .values()
463            .filter(|acc| acc.user_id == user_id)
464            .cloned()
465            .collect())
466    }
467
468    async fn delete_account(&self, id: &str) -> AuthResult<()> {
469        let mut accounts = self.accounts.lock().unwrap();
470        accounts.remove(id);
471        Ok(())
472    }
473
474    async fn create_verification(
475        &self,
476        create_verification: CreateVerification,
477    ) -> AuthResult<Verification> {
478        let mut verifications = self.verifications.lock().unwrap();
479
480        let now = Utc::now();
481        let verification = Verification {
482            id: Uuid::new_v4().to_string(),
483            identifier: create_verification.identifier,
484            value: create_verification.value.clone(),
485            expires_at: create_verification.expires_at,
486            created_at: now,
487            updated_at: now,
488        };
489
490        verifications.insert(verification.id.clone(), verification.clone());
491        Ok(verification)
492    }
493
494    async fn get_verification(
495        &self,
496        identifier: &str,
497        value: &str,
498    ) -> AuthResult<Option<Verification>> {
499        let verifications = self.verifications.lock().unwrap();
500        let now = Utc::now();
501
502        Ok(verifications
503            .values()
504            .find(|v| v.identifier == identifier && v.value == value && v.expires_at > now)
505            .cloned())
506    }
507
508    async fn get_verification_by_value(&self, value: &str) -> AuthResult<Option<Verification>> {
509        let verifications = self.verifications.lock().unwrap();
510        let now = Utc::now();
511
512        Ok(verifications
513            .values()
514            .find(|v| v.value == value && v.expires_at > now)
515            .cloned())
516    }
517
518    async fn delete_verification(&self, id: &str) -> AuthResult<()> {
519        let mut verifications = self.verifications.lock().unwrap();
520        verifications.remove(id);
521        Ok(())
522    }
523
524    async fn delete_expired_verifications(&self) -> AuthResult<usize> {
525        let mut verifications = self.verifications.lock().unwrap();
526        let now = Utc::now();
527        let initial_count = verifications.len();
528
529        verifications.retain(|_, verification| verification.expires_at > now);
530
531        Ok(initial_count - verifications.len())
532    }
533
534    // Organization operations
535    async fn create_organization(
536        &self,
537        create_org: CreateOrganization,
538    ) -> AuthResult<Organization> {
539        let mut organizations = self.organizations.lock().unwrap();
540        let mut slug_index = self.slug_index.lock().unwrap();
541
542        // Check if slug already exists
543        if slug_index.contains_key(&create_org.slug) {
544            return Err(AuthError::conflict("Organization slug already exists"));
545        }
546
547        let id = create_org.id.unwrap_or_else(|| Uuid::new_v4().to_string());
548        let now = Utc::now();
549
550        let organization = Organization {
551            id: id.clone(),
552            name: create_org.name,
553            slug: create_org.slug.clone(),
554            logo: create_org.logo,
555            metadata: create_org.metadata,
556            created_at: now,
557            updated_at: now,
558        };
559
560        organizations.insert(id.clone(), organization.clone());
561        slug_index.insert(create_org.slug, id);
562
563        Ok(organization)
564    }
565
566    async fn get_organization_by_id(&self, id: &str) -> AuthResult<Option<Organization>> {
567        let organizations = self.organizations.lock().unwrap();
568        Ok(organizations.get(id).cloned())
569    }
570
571    async fn get_organization_by_slug(&self, slug: &str) -> AuthResult<Option<Organization>> {
572        let slug_index = self.slug_index.lock().unwrap();
573        let organizations = self.organizations.lock().unwrap();
574
575        if let Some(org_id) = slug_index.get(slug) {
576            Ok(organizations.get(org_id).cloned())
577        } else {
578            Ok(None)
579        }
580    }
581
582    async fn update_organization(
583        &self,
584        id: &str,
585        update: UpdateOrganization,
586    ) -> AuthResult<Organization> {
587        let mut organizations = self.organizations.lock().unwrap();
588        let mut slug_index = self.slug_index.lock().unwrap();
589
590        let org = organizations
591            .get_mut(id)
592            .ok_or_else(|| AuthError::not_found("Organization not found"))?;
593
594        // Update slug index if slug changed
595        if let Some(new_slug) = &update.slug
596            && new_slug != &org.slug
597        {
598            // Check if new slug already exists
599            if slug_index.contains_key(new_slug) {
600                return Err(AuthError::conflict("Organization slug already exists"));
601            }
602            slug_index.remove(&org.slug);
603            slug_index.insert(new_slug.clone(), id.to_string());
604            org.slug = new_slug.clone();
605        }
606
607        if let Some(name) = update.name {
608            org.name = name;
609        }
610        if let Some(logo) = update.logo {
611            org.logo = Some(logo);
612        }
613        if let Some(metadata) = update.metadata {
614            org.metadata = Some(metadata);
615        }
616
617        org.updated_at = Utc::now();
618
619        Ok(org.clone())
620    }
621
622    async fn delete_organization(&self, id: &str) -> AuthResult<()> {
623        let mut organizations = self.organizations.lock().unwrap();
624        let mut slug_index = self.slug_index.lock().unwrap();
625        let mut members = self.members.lock().unwrap();
626        let mut invitations = self.invitations.lock().unwrap();
627
628        if let Some(org) = organizations.remove(id) {
629            slug_index.remove(&org.slug);
630        }
631
632        // Delete all related members and invitations
633        members.retain(|_, m| m.organization_id != id);
634        invitations.retain(|_, i| i.organization_id != id);
635
636        Ok(())
637    }
638
639    async fn list_user_organizations(&self, user_id: &str) -> AuthResult<Vec<Organization>> {
640        let members = self.members.lock().unwrap();
641        let organizations = self.organizations.lock().unwrap();
642
643        let org_ids: Vec<String> = members
644            .values()
645            .filter(|m| m.user_id == user_id)
646            .map(|m| m.organization_id.clone())
647            .collect();
648
649        let orgs = org_ids
650            .iter()
651            .filter_map(|id| organizations.get(id).cloned())
652            .collect();
653
654        Ok(orgs)
655    }
656
657    // Member operations
658    async fn create_member(&self, create_member: CreateMember) -> AuthResult<Member> {
659        let mut members = self.members.lock().unwrap();
660
661        // Check if member already exists
662        let exists = members.values().any(|m| {
663            m.organization_id == create_member.organization_id && m.user_id == create_member.user_id
664        });
665
666        if exists {
667            return Err(AuthError::conflict(
668                "User is already a member of this organization",
669            ));
670        }
671
672        let id = Uuid::new_v4().to_string();
673        let now = Utc::now();
674
675        let member = Member {
676            id: id.clone(),
677            organization_id: create_member.organization_id,
678            user_id: create_member.user_id,
679            role: create_member.role,
680            created_at: now,
681        };
682
683        members.insert(id, member.clone());
684
685        Ok(member)
686    }
687
688    async fn get_member(&self, organization_id: &str, user_id: &str) -> AuthResult<Option<Member>> {
689        let members = self.members.lock().unwrap();
690
691        let member = members
692            .values()
693            .find(|m| m.organization_id == organization_id && m.user_id == user_id)
694            .cloned();
695
696        Ok(member)
697    }
698
699    async fn get_member_by_id(&self, id: &str) -> AuthResult<Option<Member>> {
700        let members = self.members.lock().unwrap();
701        Ok(members.get(id).cloned())
702    }
703
704    async fn update_member_role(&self, member_id: &str, role: &str) -> AuthResult<Member> {
705        let mut members = self.members.lock().unwrap();
706
707        let member = members
708            .get_mut(member_id)
709            .ok_or_else(|| AuthError::not_found("Member not found"))?;
710
711        member.role = role.to_string();
712
713        Ok(member.clone())
714    }
715
716    async fn delete_member(&self, member_id: &str) -> AuthResult<()> {
717        let mut members = self.members.lock().unwrap();
718        members.remove(member_id);
719        Ok(())
720    }
721
722    async fn list_organization_members(&self, organization_id: &str) -> AuthResult<Vec<Member>> {
723        let members = self.members.lock().unwrap();
724
725        let org_members = members
726            .values()
727            .filter(|m| m.organization_id == organization_id)
728            .cloned()
729            .collect();
730
731        Ok(org_members)
732    }
733
734    async fn count_organization_members(&self, organization_id: &str) -> AuthResult<usize> {
735        let members = self.members.lock().unwrap();
736        let count = members
737            .values()
738            .filter(|m| m.organization_id == organization_id)
739            .count();
740        Ok(count)
741    }
742
743    async fn count_organization_owners(&self, organization_id: &str) -> AuthResult<usize> {
744        let members = self.members.lock().unwrap();
745        let count = members
746            .values()
747            .filter(|m| m.organization_id == organization_id && m.role == "owner")
748            .count();
749        Ok(count)
750    }
751
752    // Invitation operations
753    async fn create_invitation(&self, create_inv: CreateInvitation) -> AuthResult<Invitation> {
754        let mut invitations = self.invitations.lock().unwrap();
755
756        let id = Uuid::new_v4().to_string();
757        let now = Utc::now();
758
759        let invitation = Invitation {
760            id: id.clone(),
761            organization_id: create_inv.organization_id,
762            email: create_inv.email,
763            role: create_inv.role,
764            status: InvitationStatus::Pending,
765            inviter_id: create_inv.inviter_id,
766            expires_at: create_inv.expires_at,
767            created_at: now,
768        };
769
770        invitations.insert(id, invitation.clone());
771
772        Ok(invitation)
773    }
774
775    async fn get_invitation_by_id(&self, id: &str) -> AuthResult<Option<Invitation>> {
776        let invitations = self.invitations.lock().unwrap();
777        Ok(invitations.get(id).cloned())
778    }
779
780    async fn get_pending_invitation(
781        &self,
782        organization_id: &str,
783        email: &str,
784    ) -> AuthResult<Option<Invitation>> {
785        let invitations = self.invitations.lock().unwrap();
786
787        let invitation = invitations
788            .values()
789            .find(|i| {
790                i.organization_id == organization_id
791                    && i.email.to_lowercase() == email.to_lowercase()
792                    && i.status == InvitationStatus::Pending
793            })
794            .cloned();
795
796        Ok(invitation)
797    }
798
799    async fn update_invitation_status(
800        &self,
801        id: &str,
802        status: InvitationStatus,
803    ) -> AuthResult<Invitation> {
804        let mut invitations = self.invitations.lock().unwrap();
805
806        let invitation = invitations
807            .get_mut(id)
808            .ok_or_else(|| AuthError::not_found("Invitation not found"))?;
809
810        invitation.status = status;
811
812        Ok(invitation.clone())
813    }
814
815    async fn list_organization_invitations(
816        &self,
817        organization_id: &str,
818    ) -> AuthResult<Vec<Invitation>> {
819        let invitations = self.invitations.lock().unwrap();
820
821        let org_invitations = invitations
822            .values()
823            .filter(|i| i.organization_id == organization_id)
824            .cloned()
825            .collect();
826
827        Ok(org_invitations)
828    }
829
830    async fn list_user_invitations(&self, email: &str) -> AuthResult<Vec<Invitation>> {
831        let invitations = self.invitations.lock().unwrap();
832        let now = Utc::now();
833
834        let user_invitations = invitations
835            .values()
836            .filter(|i| {
837                i.email.to_lowercase() == email.to_lowercase()
838                    && i.status == InvitationStatus::Pending
839                    && i.expires_at > now
840            })
841            .cloned()
842            .collect();
843
844        Ok(user_invitations)
845    }
846
847    // Session organization support
848    async fn update_session_active_organization(
849        &self,
850        token: &str,
851        organization_id: Option<&str>,
852    ) -> AuthResult<Session> {
853        let mut sessions = self.sessions.lock().unwrap();
854
855        let session = sessions.get_mut(token).ok_or(AuthError::SessionNotFound)?;
856
857        session.active_organization_id = organization_id.map(|s| s.to_string());
858        session.updated_at = Utc::now();
859
860        Ok(session.clone())
861    }
862}
863
864#[cfg(feature = "sqlx-postgres")]
865pub mod sqlx_adapter {
866    use super::*;
867    use sqlx::PgPool;
868
869    pub struct SqlxAdapter {
870        pool: PgPool,
871    }
872
873    impl SqlxAdapter {
874        pub async fn new(database_url: &str) -> Result<Self, sqlx::Error> {
875            let pool = PgPool::connect(database_url).await?;
876            Ok(Self { pool })
877        }
878
879        /// Create adapter with custom pool configuration
880        pub async fn with_config(
881            database_url: &str,
882            config: PoolConfig,
883        ) -> Result<Self, sqlx::Error> {
884            let pool = sqlx::postgres::PgPoolOptions::new()
885                .max_connections(config.max_connections)
886                .min_connections(config.min_connections)
887                .acquire_timeout(config.acquire_timeout)
888                .idle_timeout(config.idle_timeout)
889                .max_lifetime(config.max_lifetime)
890                .connect(database_url)
891                .await?;
892            Ok(Self { pool })
893        }
894
895        pub fn from_pool(pool: PgPool) -> Self {
896            Self { pool }
897        }
898
899        /// Test database connection
900        pub async fn test_connection(&self) -> Result<(), sqlx::Error> {
901            sqlx::query("SELECT 1").execute(&self.pool).await?;
902            Ok(())
903        }
904
905        /// Get connection pool statistics
906        pub fn pool_stats(&self) -> PoolStats {
907            PoolStats {
908                size: self.pool.size(),
909                idle: self.pool.num_idle(),
910            }
911        }
912
913        /// Close the connection pool
914        pub async fn close(&self) {
915            self.pool.close().await;
916        }
917    }
918
919    /// Database connection pool configuration
920    #[derive(Debug, Clone)]
921    pub struct PoolConfig {
922        pub max_connections: u32,
923        pub min_connections: u32,
924        pub acquire_timeout: std::time::Duration,
925        pub idle_timeout: Option<std::time::Duration>,
926        pub max_lifetime: Option<std::time::Duration>,
927    }
928
929    impl Default for PoolConfig {
930        fn default() -> Self {
931            Self {
932                max_connections: 10,
933                min_connections: 0,
934                acquire_timeout: std::time::Duration::from_secs(30),
935                idle_timeout: Some(std::time::Duration::from_secs(600)), // 10 minutes
936                max_lifetime: Some(std::time::Duration::from_secs(1800)), // 30 minutes
937            }
938        }
939    }
940
941    /// Connection pool statistics
942    #[derive(Debug, Clone)]
943    pub struct PoolStats {
944        pub size: u32,
945        pub idle: usize,
946    }
947
948    #[async_trait]
949    impl DatabaseAdapter for SqlxAdapter {
950        type User = User;
951        type Session = Session;
952        type Account = Account;
953        type Organization = Organization;
954        type Member = Member;
955        type Invitation = Invitation;
956        type Verification = Verification;
957
958        async fn create_user(&self, create_user: CreateUser) -> AuthResult<User> {
959            let id = create_user.id.unwrap_or_else(|| Uuid::new_v4().to_string());
960            let now = Utc::now();
961
962            let user = sqlx::query_as::<_, User>(
963                r#"
964                INSERT INTO users (id, email, name, image, email_verified, created_at, updated_at, metadata)
965                VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
966                RETURNING id, email, name, image, email_verified, created_at, updated_at, metadata
967                "#
968            )
969            .bind(&id)
970            .bind(&create_user.email)
971            .bind(&create_user.name)
972            .bind(&create_user.image)
973            .bind(false)
974            .bind(&now)
975            .bind(&now)
976            .bind(sqlx::types::Json(create_user.metadata.unwrap_or_default()))
977            .fetch_one(&self.pool)
978            .await?;
979
980            Ok(user)
981        }
982
983        async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<User>> {
984            let user = sqlx::query_as::<_, User>(
985                r#"
986                SELECT id, email, name, image, email_verified, created_at, updated_at, metadata
987                FROM users WHERE id = $1
988                "#,
989            )
990            .bind(id)
991            .fetch_optional(&self.pool)
992            .await?;
993
994            Ok(user)
995        }
996
997        async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<User>> {
998            let user = sqlx::query_as::<_, User>(
999                r#"
1000                SELECT id, email, name, image, email_verified, created_at, updated_at, metadata
1001                FROM users WHERE email = $1
1002                "#,
1003            )
1004            .bind(email)
1005            .fetch_optional(&self.pool)
1006            .await?;
1007
1008            Ok(user)
1009        }
1010
1011        async fn get_user_by_username(&self, username: &str) -> AuthResult<Option<User>> {
1012            let user = sqlx::query_as::<_, User>(
1013                r#"
1014                SELECT id, email, name, image, email_verified, created_at, updated_at, metadata
1015                FROM users WHERE username = $1
1016                "#,
1017            )
1018            .bind(username)
1019            .fetch_optional(&self.pool)
1020            .await?;
1021
1022            Ok(user)
1023        }
1024
1025        async fn update_user(&self, id: &str, update: UpdateUser) -> AuthResult<User> {
1026            let mut query = sqlx::QueryBuilder::new("UPDATE users SET updated_at = NOW()");
1027            let mut has_updates = false;
1028
1029            if let Some(email) = &update.email {
1030                query.push(", email = ");
1031                query.push_bind(email);
1032                has_updates = true;
1033            }
1034
1035            if let Some(name) = &update.name {
1036                query.push(", name = ");
1037                query.push_bind(name);
1038                has_updates = true;
1039            }
1040
1041            if let Some(image) = &update.image {
1042                query.push(", image = ");
1043                query.push_bind(image);
1044                has_updates = true;
1045            }
1046
1047            if let Some(email_verified) = update.email_verified {
1048                query.push(", email_verified = ");
1049                query.push_bind(email_verified);
1050                has_updates = true;
1051            }
1052
1053            if let Some(metadata) = &update.metadata {
1054                query.push(", metadata = ");
1055                query.push_bind(sqlx::types::Json(metadata.clone()));
1056                has_updates = true;
1057            }
1058
1059            if !has_updates {
1060                // If no updates, just return the current user
1061                return self
1062                    .get_user_by_id(id)
1063                    .await?
1064                    .ok_or(AuthError::UserNotFound);
1065            }
1066
1067            query.push(" WHERE id = ");
1068            query.push_bind(id);
1069            query.push(" RETURNING id, email, name, image, email_verified, created_at, updated_at, metadata");
1070
1071            let user = query.build_query_as::<User>().fetch_one(&self.pool).await?;
1072
1073            Ok(user)
1074        }
1075
1076        async fn delete_user(&self, id: &str) -> AuthResult<()> {
1077            sqlx::query("DELETE FROM users WHERE id = $1")
1078                .bind(id)
1079                .execute(&self.pool)
1080                .await?;
1081
1082            Ok(())
1083        }
1084
1085        async fn create_session(&self, create_session: CreateSession) -> AuthResult<Session> {
1086            let id = Uuid::new_v4().to_string();
1087            let token = format!("session_{}", Uuid::new_v4());
1088            let now = Utc::now();
1089
1090            let session = sqlx::query_as::<_, Session>(
1091                r#"
1092                INSERT INTO sessions (id, user_id, token, expires_at, created_at, ip_address, user_agent, active)
1093                VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
1094                RETURNING id, user_id, token, expires_at, created_at, ip_address, user_agent, active
1095                "#
1096            )
1097            .bind(&id)
1098            .bind(&create_session.user_id)
1099            .bind(&token)
1100            .bind(&create_session.expires_at)
1101            .bind(&now)
1102            .bind(&create_session.ip_address)
1103            .bind(&create_session.user_agent)
1104            .bind(true)
1105            .fetch_one(&self.pool)
1106            .await?;
1107
1108            Ok(session)
1109        }
1110
1111        async fn get_session(&self, token: &str) -> AuthResult<Option<Session>> {
1112            let session = sqlx::query_as::<_, Session>(
1113                r#"
1114                SELECT id, user_id, token, expires_at, created_at, updated_at, ip_address, user_agent, active, impersonated_by, active_organization_id
1115                FROM sessions 
1116                WHERE token = $1 AND active = true
1117                "#
1118            )
1119            .bind(token)
1120            .fetch_optional(&self.pool)
1121            .await?;
1122
1123            Ok(session)
1124        }
1125
1126        async fn get_user_sessions(&self, user_id: &str) -> AuthResult<Vec<Session>> {
1127            let sessions = sqlx::query_as::<_, Session>(
1128                r#"
1129                SELECT id, user_id, token, expires_at, created_at, updated_at, ip_address, user_agent, active, impersonated_by, active_organization_id
1130                FROM sessions 
1131                WHERE user_id = $1 AND active = true
1132                ORDER BY created_at DESC
1133                "#
1134            )
1135            .bind(user_id)
1136            .fetch_all(&self.pool)
1137            .await?;
1138
1139            Ok(sessions)
1140        }
1141
1142        async fn update_session_expiry(
1143            &self,
1144            token: &str,
1145            expires_at: DateTime<Utc>,
1146        ) -> AuthResult<()> {
1147            sqlx::query(
1148                r#"
1149                UPDATE sessions 
1150                SET expires_at = $1 
1151                WHERE token = $2 AND active = true
1152                "#,
1153            )
1154            .bind(&expires_at)
1155            .bind(token)
1156            .execute(&self.pool)
1157            .await?;
1158
1159            Ok(())
1160        }
1161
1162        async fn delete_session(&self, token: &str) -> AuthResult<()> {
1163            sqlx::query("DELETE FROM sessions WHERE token = $1")
1164                .bind(token)
1165                .execute(&self.pool)
1166                .await?;
1167
1168            Ok(())
1169        }
1170
1171        async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()> {
1172            sqlx::query("DELETE FROM sessions WHERE user_id = $1")
1173                .bind(user_id)
1174                .execute(&self.pool)
1175                .await?;
1176
1177            Ok(())
1178        }
1179
1180        async fn delete_expired_sessions(&self) -> AuthResult<usize> {
1181            let result =
1182                sqlx::query("DELETE FROM sessions WHERE expires_at < NOW() OR active = false")
1183                    .execute(&self.pool)
1184                    .await?;
1185
1186            Ok(result.rows_affected() as usize)
1187        }
1188
1189        async fn create_account(&self, create_account: CreateAccount) -> AuthResult<Account> {
1190            let id = Uuid::new_v4().to_string();
1191            let now = Utc::now();
1192
1193            let account = sqlx::query_as::<_, Account>(
1194                r#"
1195                INSERT INTO accounts (id, account_id, provider_id, user_id, access_token, refresh_token, id_token, access_token_expires_at, refresh_token_expires_at, scope, password, created_at, updated_at)
1196                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
1197                RETURNING *
1198                "#
1199            )
1200            .bind(&id)
1201            .bind(&create_account.account_id)
1202            .bind(&create_account.provider_id)
1203            .bind(&create_account.user_id)
1204            .bind(&create_account.access_token)
1205            .bind(&create_account.refresh_token)
1206            .bind(&create_account.id_token)
1207            .bind(&create_account.access_token_expires_at)
1208            .bind(&create_account.refresh_token_expires_at)
1209            .bind(&create_account.scope)
1210            .bind(&create_account.password)
1211            .bind(&now)
1212            .bind(&now)
1213            .fetch_one(&self.pool)
1214            .await?;
1215
1216            Ok(account)
1217        }
1218
1219        async fn get_account(
1220            &self,
1221            provider: &str,
1222            provider_account_id: &str,
1223        ) -> AuthResult<Option<Account>> {
1224            let account = sqlx::query_as::<_, Account>(
1225                r#"
1226                SELECT *
1227                FROM accounts 
1228                WHERE provider_id = $1 AND account_id = $2
1229                "#,
1230            )
1231            .bind(provider)
1232            .bind(provider_account_id)
1233            .fetch_optional(&self.pool)
1234            .await?;
1235
1236            Ok(account)
1237        }
1238
1239        async fn get_user_accounts(&self, user_id: &str) -> AuthResult<Vec<Account>> {
1240            let accounts = sqlx::query_as::<_, Account>(
1241                r#"
1242                SELECT id, user_id, provider, provider_account_id, access_token, refresh_token, expires_at, token_type, scope, created_at
1243                FROM accounts 
1244                WHERE user_id = $1
1245                ORDER BY created_at DESC
1246                "#
1247            )
1248            .bind(user_id)
1249            .fetch_all(&self.pool)
1250            .await?;
1251
1252            Ok(accounts)
1253        }
1254
1255        async fn delete_account(&self, id: &str) -> AuthResult<()> {
1256            sqlx::query("DELETE FROM accounts WHERE id = $1")
1257                .bind(id)
1258                .execute(&self.pool)
1259                .await?;
1260
1261            Ok(())
1262        }
1263
1264        async fn create_verification(
1265            &self,
1266            create_verification: CreateVerification,
1267        ) -> AuthResult<Verification> {
1268            let id = Uuid::new_v4().to_string();
1269            let now = Utc::now();
1270
1271            let verification = sqlx::query_as::<_, Verification>(
1272                r#"
1273                INSERT INTO verifications (id, identifier, value, expires_at, created_at, updated_at)
1274                VALUES ($1, $2, $3, $4, $5, $6)
1275                RETURNING *
1276                "#
1277            )
1278            .bind(&id)
1279            .bind(&create_verification.identifier)
1280            .bind(&create_verification.value)
1281            .bind(&create_verification.expires_at)
1282            .bind(&now)
1283            .bind(&now)
1284            .fetch_one(&self.pool)
1285            .await?;
1286
1287            Ok(verification)
1288        }
1289
1290        async fn get_verification(
1291            &self,
1292            identifier: &str,
1293            value: &str,
1294        ) -> AuthResult<Option<Verification>> {
1295            let verification = sqlx::query_as::<_, Verification>(
1296                r#"
1297                SELECT *
1298                FROM verifications 
1299                WHERE identifier = $1 AND value = $2 AND expires_at > NOW()
1300                "#,
1301            )
1302            .bind(identifier)
1303            .bind(value)
1304            .fetch_optional(&self.pool)
1305            .await?;
1306
1307            Ok(verification)
1308        }
1309
1310        async fn get_verification_by_value(&self, value: &str) -> AuthResult<Option<Verification>> {
1311            let verification = sqlx::query_as::<_, Verification>(
1312                r#"
1313                SELECT *
1314                FROM verifications 
1315                WHERE value = $1 AND expires_at > NOW()
1316                "#,
1317            )
1318            .bind(value)
1319            .fetch_optional(&self.pool)
1320            .await?;
1321
1322            Ok(verification)
1323        }
1324
1325        async fn delete_verification(&self, id: &str) -> AuthResult<()> {
1326            sqlx::query("DELETE FROM verifications WHERE id = $1")
1327                .bind(id)
1328                .execute(&self.pool)
1329                .await?;
1330
1331            Ok(())
1332        }
1333
1334        async fn delete_expired_verifications(&self) -> AuthResult<usize> {
1335            let result = sqlx::query("DELETE FROM verifications WHERE expires_at < NOW()")
1336                .execute(&self.pool)
1337                .await?;
1338
1339            Ok(result.rows_affected() as usize)
1340        }
1341
1342        // Organization operations
1343        async fn create_organization(
1344            &self,
1345            create_org: CreateOrganization,
1346        ) -> AuthResult<Organization> {
1347            let id = create_org.id.unwrap_or_else(|| Uuid::new_v4().to_string());
1348            let now = Utc::now();
1349
1350            let organization = sqlx::query_as::<_, Organization>(
1351                r#"
1352                INSERT INTO organization (id, name, slug, logo, metadata, created_at, updated_at)
1353                VALUES ($1, $2, $3, $4, $5, $6, $7)
1354                RETURNING id, name, slug, logo, metadata, created_at, updated_at
1355                "#,
1356            )
1357            .bind(&id)
1358            .bind(&create_org.name)
1359            .bind(&create_org.slug)
1360            .bind(&create_org.logo)
1361            .bind(sqlx::types::Json(
1362                create_org.metadata.unwrap_or(serde_json::json!({})),
1363            ))
1364            .bind(&now)
1365            .bind(&now)
1366            .fetch_one(&self.pool)
1367            .await?;
1368
1369            Ok(organization)
1370        }
1371
1372        async fn get_organization_by_id(&self, id: &str) -> AuthResult<Option<Organization>> {
1373            let organization = sqlx::query_as::<_, Organization>(
1374                r#"
1375                SELECT id, name, slug, logo, metadata, created_at, updated_at
1376                FROM organization WHERE id = $1
1377                "#,
1378            )
1379            .bind(id)
1380            .fetch_optional(&self.pool)
1381            .await?;
1382
1383            Ok(organization)
1384        }
1385
1386        async fn get_organization_by_slug(&self, slug: &str) -> AuthResult<Option<Organization>> {
1387            let organization = sqlx::query_as::<_, Organization>(
1388                r#"
1389                SELECT id, name, slug, logo, metadata, created_at, updated_at
1390                FROM organization WHERE slug = $1
1391                "#,
1392            )
1393            .bind(slug)
1394            .fetch_optional(&self.pool)
1395            .await?;
1396
1397            Ok(organization)
1398        }
1399
1400        async fn update_organization(
1401            &self,
1402            id: &str,
1403            update: UpdateOrganization,
1404        ) -> AuthResult<Organization> {
1405            let mut query = sqlx::QueryBuilder::new("UPDATE organization SET updated_at = NOW()");
1406
1407            if let Some(name) = &update.name {
1408                query.push(", name = ");
1409                query.push_bind(name);
1410            }
1411            if let Some(slug) = &update.slug {
1412                query.push(", slug = ");
1413                query.push_bind(slug);
1414            }
1415            if let Some(logo) = &update.logo {
1416                query.push(", logo = ");
1417                query.push_bind(logo);
1418            }
1419            if let Some(metadata) = &update.metadata {
1420                query.push(", metadata = ");
1421                query.push_bind(sqlx::types::Json(metadata.clone()));
1422            }
1423
1424            query.push(" WHERE id = ");
1425            query.push_bind(id);
1426            query.push(" RETURNING id, name, slug, logo, metadata, created_at, updated_at");
1427
1428            let organization = query
1429                .build_query_as::<Organization>()
1430                .fetch_one(&self.pool)
1431                .await?;
1432
1433            Ok(organization)
1434        }
1435
1436        async fn delete_organization(&self, id: &str) -> AuthResult<()> {
1437            sqlx::query("DELETE FROM organization WHERE id = $1")
1438                .bind(id)
1439                .execute(&self.pool)
1440                .await?;
1441
1442            Ok(())
1443        }
1444
1445        async fn list_user_organizations(&self, user_id: &str) -> AuthResult<Vec<Organization>> {
1446            let organizations = sqlx::query_as::<_, Organization>(
1447                r#"
1448                SELECT o.id, o.name, o.slug, o.logo, o.metadata, o.created_at, o.updated_at
1449                FROM organization o
1450                INNER JOIN member m ON o.id = m.organization_id
1451                WHERE m.user_id = $1
1452                ORDER BY o.created_at DESC
1453                "#,
1454            )
1455            .bind(user_id)
1456            .fetch_all(&self.pool)
1457            .await?;
1458
1459            Ok(organizations)
1460        }
1461
1462        // Member operations
1463        async fn create_member(&self, create_member: CreateMember) -> AuthResult<Member> {
1464            let id = Uuid::new_v4().to_string();
1465            let now = Utc::now();
1466
1467            let member = sqlx::query_as::<_, Member>(
1468                r#"
1469                INSERT INTO member (id, organization_id, user_id, role, created_at)
1470                VALUES ($1, $2, $3, $4, $5)
1471                RETURNING id, organization_id, user_id, role, created_at
1472                "#,
1473            )
1474            .bind(&id)
1475            .bind(&create_member.organization_id)
1476            .bind(&create_member.user_id)
1477            .bind(&create_member.role)
1478            .bind(&now)
1479            .fetch_one(&self.pool)
1480            .await?;
1481
1482            Ok(member)
1483        }
1484
1485        async fn get_member(
1486            &self,
1487            organization_id: &str,
1488            user_id: &str,
1489        ) -> AuthResult<Option<Member>> {
1490            let member = sqlx::query_as::<_, Member>(
1491                r#"
1492                SELECT id, organization_id, user_id, role, created_at
1493                FROM member
1494                WHERE organization_id = $1 AND user_id = $2
1495                "#,
1496            )
1497            .bind(organization_id)
1498            .bind(user_id)
1499            .fetch_optional(&self.pool)
1500            .await?;
1501
1502            Ok(member)
1503        }
1504
1505        async fn get_member_by_id(&self, id: &str) -> AuthResult<Option<Member>> {
1506            let member = sqlx::query_as::<_, Member>(
1507                r#"
1508                SELECT id, organization_id, user_id, role, created_at
1509                FROM member
1510                WHERE id = $1
1511                "#,
1512            )
1513            .bind(id)
1514            .fetch_optional(&self.pool)
1515            .await?;
1516
1517            Ok(member)
1518        }
1519
1520        async fn update_member_role(&self, member_id: &str, role: &str) -> AuthResult<Member> {
1521            let member = sqlx::query_as::<_, Member>(
1522                r#"
1523                UPDATE member SET role = $1
1524                WHERE id = $2
1525                RETURNING id, organization_id, user_id, role, created_at
1526                "#,
1527            )
1528            .bind(role)
1529            .bind(member_id)
1530            .fetch_one(&self.pool)
1531            .await?;
1532
1533            Ok(member)
1534        }
1535
1536        async fn delete_member(&self, member_id: &str) -> AuthResult<()> {
1537            sqlx::query("DELETE FROM member WHERE id = $1")
1538                .bind(member_id)
1539                .execute(&self.pool)
1540                .await?;
1541
1542            Ok(())
1543        }
1544
1545        async fn list_organization_members(
1546            &self,
1547            organization_id: &str,
1548        ) -> AuthResult<Vec<Member>> {
1549            let members = sqlx::query_as::<_, Member>(
1550                r#"
1551                SELECT id, organization_id, user_id, role, created_at
1552                FROM member
1553                WHERE organization_id = $1
1554                ORDER BY created_at ASC
1555                "#,
1556            )
1557            .bind(organization_id)
1558            .fetch_all(&self.pool)
1559            .await?;
1560
1561            Ok(members)
1562        }
1563
1564        async fn count_organization_members(&self, organization_id: &str) -> AuthResult<usize> {
1565            let count: (i64,) =
1566                sqlx::query_as("SELECT COUNT(*) FROM member WHERE organization_id = $1")
1567                    .bind(organization_id)
1568                    .fetch_one(&self.pool)
1569                    .await?;
1570
1571            Ok(count.0 as usize)
1572        }
1573
1574        async fn count_organization_owners(&self, organization_id: &str) -> AuthResult<usize> {
1575            let count: (i64,) = sqlx::query_as(
1576                "SELECT COUNT(*) FROM member WHERE organization_id = $1 AND role = 'owner'",
1577            )
1578            .bind(organization_id)
1579            .fetch_one(&self.pool)
1580            .await?;
1581
1582            Ok(count.0 as usize)
1583        }
1584
1585        // Invitation operations
1586        async fn create_invitation(&self, create_inv: CreateInvitation) -> AuthResult<Invitation> {
1587            let id = Uuid::new_v4().to_string();
1588            let now = Utc::now();
1589
1590            let invitation = sqlx::query_as::<_, Invitation>(
1591                r#"
1592                INSERT INTO invitation (id, organization_id, email, role, status, inviter_id, expires_at, created_at)
1593                VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7)
1594                RETURNING id, organization_id, email, role, status, inviter_id, expires_at, created_at
1595                "#,
1596            )
1597            .bind(&id)
1598            .bind(&create_inv.organization_id)
1599            .bind(&create_inv.email)
1600            .bind(&create_inv.role)
1601            .bind(&create_inv.inviter_id)
1602            .bind(&create_inv.expires_at)
1603            .bind(&now)
1604            .fetch_one(&self.pool)
1605            .await?;
1606
1607            Ok(invitation)
1608        }
1609
1610        async fn get_invitation_by_id(&self, id: &str) -> AuthResult<Option<Invitation>> {
1611            let invitation = sqlx::query_as::<_, Invitation>(
1612                r#"
1613                SELECT id, organization_id, email, role, status, inviter_id, expires_at, created_at
1614                FROM invitation
1615                WHERE id = $1
1616                "#,
1617            )
1618            .bind(id)
1619            .fetch_optional(&self.pool)
1620            .await?;
1621
1622            Ok(invitation)
1623        }
1624
1625        async fn get_pending_invitation(
1626            &self,
1627            organization_id: &str,
1628            email: &str,
1629        ) -> AuthResult<Option<Invitation>> {
1630            let invitation = sqlx::query_as::<_, Invitation>(
1631                r#"
1632                SELECT id, organization_id, email, role, status, inviter_id, expires_at, created_at
1633                FROM invitation
1634                WHERE organization_id = $1 AND LOWER(email) = LOWER($2) AND status = 'pending'
1635                "#,
1636            )
1637            .bind(organization_id)
1638            .bind(email)
1639            .fetch_optional(&self.pool)
1640            .await?;
1641
1642            Ok(invitation)
1643        }
1644
1645        async fn update_invitation_status(
1646            &self,
1647            id: &str,
1648            status: InvitationStatus,
1649        ) -> AuthResult<Invitation> {
1650            let invitation = sqlx::query_as::<_, Invitation>(
1651                r#"
1652                UPDATE invitation SET status = $1
1653                WHERE id = $2
1654                RETURNING id, organization_id, email, role, status, inviter_id, expires_at, created_at
1655                "#,
1656            )
1657            .bind(status.to_string())
1658            .bind(id)
1659            .fetch_one(&self.pool)
1660            .await?;
1661
1662            Ok(invitation)
1663        }
1664
1665        async fn list_organization_invitations(
1666            &self,
1667            organization_id: &str,
1668        ) -> AuthResult<Vec<Invitation>> {
1669            let invitations = sqlx::query_as::<_, Invitation>(
1670                r#"
1671                SELECT id, organization_id, email, role, status, inviter_id, expires_at, created_at
1672                FROM invitation
1673                WHERE organization_id = $1
1674                ORDER BY created_at DESC
1675                "#,
1676            )
1677            .bind(organization_id)
1678            .fetch_all(&self.pool)
1679            .await?;
1680
1681            Ok(invitations)
1682        }
1683
1684        async fn list_user_invitations(&self, email: &str) -> AuthResult<Vec<Invitation>> {
1685            let invitations = sqlx::query_as::<_, Invitation>(
1686                r#"
1687                SELECT id, organization_id, email, role, status, inviter_id, expires_at, created_at
1688                FROM invitation
1689                WHERE LOWER(email) = LOWER($1) AND status = 'pending' AND expires_at > NOW()
1690                ORDER BY created_at DESC
1691                "#,
1692            )
1693            .bind(email)
1694            .fetch_all(&self.pool)
1695            .await?;
1696
1697            Ok(invitations)
1698        }
1699
1700        // Session organization support
1701        async fn update_session_active_organization(
1702            &self,
1703            token: &str,
1704            organization_id: Option<&str>,
1705        ) -> AuthResult<Session> {
1706            let session = sqlx::query_as::<_, Session>(
1707                r#"
1708                UPDATE sessions SET active_organization_id = $1, updated_at = NOW()
1709                WHERE token = $2 AND active = true
1710                RETURNING id, user_id, token, expires_at, created_at, updated_at, ip_address, user_agent, active, impersonated_by, active_organization_id
1711                "#,
1712            )
1713            .bind(organization_id)
1714            .bind(token)
1715            .fetch_one(&self.pool)
1716            .await?;
1717
1718            Ok(session)
1719        }
1720    }
1721}
1722
1723#[cfg(feature = "sqlx-postgres")]
1724pub use sqlx_adapter::SqlxAdapter;
better_auth_core/adapters/database.rs

better_auth_core/adapters/
database.rs
