kcode_k1_groups_projection/
lib.rs

1use std::{
2    collections::{HashMap, HashSet},
3    fs,
4    hash::Hash,
5    path::{Path, PathBuf},
6    sync::{
7        Mutex, RwLock,
8        atomic::{AtomicBool, Ordering},
9    },
10    time::{Duration, Instant},
11};
12
13pub use kcode_k1_invites::UserId;
14use kcode_k1_transaction::{SubsystemId, Transaction};
15use kcode_k1_txn_ordering::K1TxnOrdering;
16pub use kcode_k1_txn_ordering::TxId;
17use rusqlite::{Connection, Error as SqlError, TransactionBehavior, params};
18
19const DATABASE_NAME: &str = "groups.sqlite3";
20const SUBSYSTEM: &str = "k1-groups-subsystem";
21const CREATE_METADATA: &str = "CREATE TABLE metadata (singleton INTEGER PRIMARY KEY CHECK (singleton = 1), schema_version INTEGER NOT NULL, last_applied_txid BLOB)";
22const CREATE_GROUPS: &str =
23    "CREATE TABLE groups (group_id BLOB PRIMARY KEY, revision BLOB NOT NULL)";
24const CREATE_USERS: &str = "CREATE TABLE user_memberships (group_id BLOB NOT NULL, user_id BLOB NOT NULL, role INTEGER NOT NULL, PRIMARY KEY (group_id, user_id), FOREIGN KEY (group_id) REFERENCES groups(group_id) ON DELETE CASCADE)";
25const CREATE_MODELS: &str = "CREATE TABLE model_memberships (group_id BLOB NOT NULL, model_id BLOB NOT NULL, PRIMARY KEY (group_id, model_id), FOREIGN KEY (group_id) REFERENCES groups(group_id) ON DELETE CASCADE)";
26const CREATE_USER_INDEX: &str =
27    "CREATE INDEX user_memberships_by_user ON user_memberships (user_id, group_id)";
28const CREATE_MODEL_INDEX: &str =
29    "CREATE INDEX model_memberships_by_model ON model_memberships (model_id, group_id)";
30
31#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
32pub struct GroupId(TxId);
33
34impl GroupId {
35    pub const fn new(txid: TxId) -> Self {
36        Self(txid)
37    }
38
39    pub const fn txid(self) -> TxId {
40        self.0
41    }
42}
43
44#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
45pub struct ModelId([u8; 32]);
46
47impl ModelId {
48    pub const fn from_bytes(bytes: [u8; 32]) -> Self {
49        Self(bytes)
50    }
51
52    pub const fn as_bytes(&self) -> &[u8; 32] {
53        &self.0
54    }
55
56    pub const fn into_bytes(self) -> [u8; 32] {
57        self.0
58    }
59}
60
61#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
62pub enum GroupRole {
63    User,
64    Admin,
65    Owner,
66}
67
68impl GroupRole {
69    fn stored(self) -> i64 {
70        match self {
71            Self::User => 0,
72            Self::Admin => 1,
73            Self::Owner => 2,
74        }
75    }
76
77    fn from_stored(value: i64) -> Option<Self> {
78        match value {
79            0 => Some(Self::User),
80            1 => Some(Self::Admin),
81            2 => Some(Self::Owner),
82            _ => None,
83        }
84    }
85}
86
87#[derive(Clone, Debug, Eq, PartialEq)]
88pub struct GroupUser {
89    user_id: UserId,
90    role: GroupRole,
91}
92
93impl GroupUser {
94    pub fn user_id(&self) -> UserId {
95        self.user_id
96    }
97
98    pub fn role(&self) -> GroupRole {
99        self.role
100    }
101}
102
103#[derive(Clone, Debug, Eq, PartialEq)]
104pub struct GroupRevision {
105    group_id: GroupId,
106    txid: TxId,
107}
108
109impl GroupRevision {
110    pub fn group_id(&self) -> GroupId {
111        self.group_id
112    }
113
114    pub fn txid(&self) -> TxId {
115        self.txid
116    }
117}
118
119#[derive(Clone, Debug, Eq, PartialEq)]
120pub struct Group {
121    id: GroupId,
122    revision: GroupRevision,
123    users: Vec<GroupUser>,
124    models: Vec<ModelId>,
125}
126
127impl Group {
128    pub fn id(&self) -> GroupId {
129        self.id
130    }
131
132    pub fn revision(&self) -> &GroupRevision {
133        &self.revision
134    }
135
136    pub fn users(&self) -> &[GroupUser] {
137        &self.users
138    }
139
140    pub fn models(&self) -> &[ModelId] {
141        &self.models
142    }
143}
144
145#[derive(Clone, Debug, Eq, PartialEq)]
146pub struct GroupMemberships {
147    revision: Option<TxId>,
148    user_groups: Vec<GroupId>,
149    model_groups: Vec<GroupId>,
150    shared_groups: Vec<GroupId>,
151}
152
153impl GroupMemberships {
154    pub fn revision(&self) -> Option<TxId> {
155        self.revision
156    }
157
158    pub fn user_groups(&self) -> &[GroupId] {
159        &self.user_groups
160    }
161
162    pub fn model_groups(&self) -> &[GroupId] {
163        &self.model_groups
164    }
165
166    pub fn shared_groups(&self) -> &[GroupId] {
167        &self.shared_groups
168    }
169}
170
171#[derive(Clone, Debug, Eq, PartialEq)]
172pub enum GroupAction {
173    Create {
174        owner: UserId,
175    },
176    SetUserRole {
177        group: GroupId,
178        actor: UserId,
179        user: UserId,
180        role: Option<GroupRole>,
181    },
182    SetModelMembership {
183        group: GroupId,
184        actor: UserId,
185        model: ModelId,
186        present: bool,
187    },
188}
189
190#[derive(Clone, Debug, Eq, PartialEq)]
191pub enum ApplyOutcome {
192    Applied(GroupRevision),
193    Unchanged(GroupRevision),
194    Rejected(String),
195}
196
197#[derive(Default)]
198struct State {
199    cursor: Option<TxId>,
200    groups: HashMap<GroupId, GroupState>,
201    user_groups: HashMap<UserId, HashSet<GroupId>>,
202    model_groups: HashMap<ModelId, HashSet<GroupId>>,
203}
204
205struct GroupState {
206    revision: TxId,
207    users: HashMap<UserId, GroupRole>,
208    models: HashSet<ModelId>,
209    owners: usize,
210}
211
212pub struct Projection {
213    state: RwLock<State>,
214    apply_lane: Mutex<()>,
215    connection: Mutex<Connection>,
216    unavailable: AtomicBool,
217}
218
219impl Projection {
220    pub fn open(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, Option<TxId>), String> {
221        let started = Instant::now();
222        let result = Self::open_inner(root, ordering);
223        let elapsed = started.elapsed();
224        if elapsed > Duration::from_millis(100) {
225            let outcome = if result.is_ok() { "ready" } else { "error" };
226            eprintln!(
227                "level=warn module=kcode-k1-groups-projection operation=open elapsed_us={} outcome={outcome}",
228                elapsed.as_micros()
229            );
230        }
231        result
232    }
233
234    fn open_inner(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, Option<TxId>), String> {
235        fs::create_dir_all(root)
236            .map_err(|error| format!("create projection root {}: {error}", root.display()))?;
237        let database = root.join(DATABASE_NAME);
238        let (connection, state) = load_or_rebuild(&database, ordering)?;
239        let cursor = state.cursor;
240        Ok((
241            Self {
242                state: RwLock::new(state),
243                apply_lane: Mutex::new(()),
244                connection: Mutex::new(connection),
245                unavailable: AtomicBool::new(false),
246            },
247            cursor,
248        ))
249    }
250
251    pub fn apply(&self, callback_txid: TxId, action: GroupAction) -> Result<ApplyOutcome, String> {
252        self.ensure_available()?;
253        let _lane = self.apply_lane.lock().map_err(|_| {
254            self.mark_unavailable();
255            "projection apply lane is unavailable until reopen".to_owned()
256        })?;
257        self.ensure_available()?;
258        let prepared = self.prepare(callback_txid, action)?;
259        {
260            let mut connection = self.connection.lock().map_err(|_| {
261                self.mark_unavailable();
262                "projection database lane is unavailable until reopen".to_owned()
263            })?;
264            if let Err(error) = persist_prepared(&mut connection, callback_txid, &prepared) {
265                self.mark_unavailable();
266                return Err(error);
267            }
268        }
269        self.publish(callback_txid, prepared)
270    }
271
272    pub fn get(&self, group: GroupId) -> Result<Option<Group>, String> {
273        self.ensure_available()?;
274        let state = self.state.read().map_err(|_| {
275            self.mark_unavailable();
276            "projection state is unavailable until reopen".to_owned()
277        })?;
278        self.ensure_available()?;
279        let Some(stored) = state.groups.get(&group) else {
280            return Ok(None);
281        };
282        let mut users = Vec::new();
283        users
284            .try_reserve(stored.users.len())
285            .map_err(|error| format!("reserve group users: {error}"))?;
286        users.extend(stored.users.iter().map(|(user_id, role)| GroupUser {
287            user_id: *user_id,
288            role: *role,
289        }));
290        let mut models = Vec::new();
291        models
292            .try_reserve(stored.models.len())
293            .map_err(|error| format!("reserve group models: {error}"))?;
294        models.extend(stored.models.iter().copied());
295        Ok(Some(Group {
296            id: group,
297            revision: GroupRevision {
298                group_id: group,
299                txid: stored.revision,
300            },
301            users,
302            models,
303        }))
304    }
305
306    pub fn groups_for_user(&self, user: UserId) -> Result<Vec<GroupId>, String> {
307        self.ensure_available()?;
308        let state = self.state.read().map_err(|_| {
309            self.mark_unavailable();
310            "projection state is unavailable until reopen".to_owned()
311        })?;
312        self.ensure_available()?;
313        copy_set(state.user_groups.get(&user), "user groups")
314    }
315
316    pub fn groups_for_model(&self, model: ModelId) -> Result<Vec<GroupId>, String> {
317        self.ensure_available()?;
318        let state = self.state.read().map_err(|_| {
319            self.mark_unavailable();
320            "projection state is unavailable until reopen".to_owned()
321        })?;
322        self.ensure_available()?;
323        copy_set(state.model_groups.get(&model), "model groups")
324    }
325
326    pub fn memberships(&self, user: UserId, model: ModelId) -> Result<GroupMemberships, String> {
327        self.ensure_available()?;
328        let state = self.state.read().map_err(|_| {
329            self.mark_unavailable();
330            "projection state is unavailable until reopen".to_owned()
331        })?;
332        self.ensure_available()?;
333        let user_set = state.user_groups.get(&user);
334        let model_set = state.model_groups.get(&model);
335        let user_groups = copy_set(user_set, "membership user groups")?;
336        let model_groups = copy_set(model_set, "membership model groups")?;
337        let mut shared_groups = Vec::new();
338        let shared_capacity = user_set
339            .map(HashSet::len)
340            .unwrap_or_default()
341            .min(model_set.map(HashSet::len).unwrap_or_default());
342        shared_groups
343            .try_reserve(shared_capacity)
344            .map_err(|error| format!("reserve shared groups: {error}"))?;
345        if let (Some(users), Some(models)) = (user_set, model_set) {
346            let (smaller, larger) = if users.len() <= models.len() {
347                (users, models)
348            } else {
349                (models, users)
350            };
351            shared_groups.extend(
352                smaller
353                    .iter()
354                    .filter(|group| larger.contains(group))
355                    .copied(),
356            );
357        }
358        Ok(GroupMemberships {
359            revision: state.cursor,
360            user_groups,
361            model_groups,
362            shared_groups,
363        })
364    }
365
366    pub fn clear(&self) -> Result<(), String> {
367        self.ensure_available()?;
368        let _lane = self.apply_lane.lock().map_err(|_| {
369            self.mark_unavailable();
370            "projection apply lane is unavailable until reopen".to_owned()
371        })?;
372        self.ensure_available()?;
373        {
374            let mut connection = self.connection.lock().map_err(|_| {
375                self.mark_unavailable();
376                "projection database lane is unavailable until reopen".to_owned()
377            })?;
378            if let Err(error) = persist_clear(&mut connection) {
379                self.mark_unavailable();
380                return Err(error);
381            }
382        }
383        let old_state = {
384            let mut state = self.state.write().map_err(|_| {
385                self.mark_unavailable();
386                "projection state is unavailable until reopen".to_owned()
387            })?;
388            std::mem::take(&mut *state)
389        };
390        drop(old_state);
391        Ok(())
392    }
393
394    fn prepare(&self, callback_txid: TxId, action: GroupAction) -> Result<Prepared, String> {
395        let mut state = self.state.write().map_err(|_| {
396            self.mark_unavailable();
397            "projection state is unavailable until reopen".to_owned()
398        })?;
399        self.ensure_available()?;
400        match action {
401            GroupAction::Create { owner } => {
402                let group = GroupId::new(callback_txid);
403                if state.groups.contains_key(&group) {
404                    return Ok(Prepared::rejected("group already exists"));
405                }
406                state
407                    .groups
408                    .try_reserve(1)
409                    .map_err(|error| format!("reserve group index: {error}"))?;
410                let new_reverse = if let Some(reverse) = state.user_groups.get_mut(&owner) {
411                    reverse
412                        .try_reserve(1)
413                        .map_err(|error| format!("reserve owner reverse membership: {error}"))?;
414                    None
415                } else {
416                    state
417                        .user_groups
418                        .try_reserve(1)
419                        .map_err(|error| format!("reserve user reverse index: {error}"))?;
420                    let mut reverse = HashSet::new();
421                    reverse
422                        .try_reserve(1)
423                        .map_err(|error| format!("reserve owner reverse membership: {error}"))?;
424                    reverse.insert(group);
425                    Some(reverse)
426                };
427                let mut users = HashMap::new();
428                users
429                    .try_reserve(1)
430                    .map_err(|error| format!("reserve owner membership: {error}"))?;
431                users.insert(owner, GroupRole::Owner);
432                Ok(Prepared {
433                    result: PreparedResult::Applied(GroupRevision {
434                        group_id: group,
435                        txid: callback_txid,
436                    }),
437                    mutation: Mutation::Create {
438                        group,
439                        owner,
440                        stored: GroupState {
441                            revision: callback_txid,
442                            users,
443                            models: HashSet::new(),
444                            owners: 1,
445                        },
446                        new_reverse,
447                    },
448                })
449            }
450            GroupAction::SetUserRole {
451                group,
452                actor,
453                user,
454                role,
455            } => prepare_user_change(&mut state, group, actor, user, role, callback_txid),
456            GroupAction::SetModelMembership {
457                group,
458                actor,
459                model,
460                present,
461            } => prepare_model_change(&mut state, group, actor, model, present, callback_txid),
462        }
463    }
464
465    fn publish(&self, callback_txid: TxId, prepared: Prepared) -> Result<ApplyOutcome, String> {
466        let mut state = self.state.write().map_err(|_| {
467            self.mark_unavailable();
468            "projection state is unavailable until reopen".to_owned()
469        })?;
470        if !mutation_matches(&state, &prepared.mutation) {
471            self.mark_unavailable();
472            return Err(
473                "projection state contradicted committed database state; reopen required"
474                    .to_owned(),
475            );
476        }
477        match prepared.mutation {
478            Mutation::None => {}
479            Mutation::Create {
480                group,
481                owner,
482                stored,
483                new_reverse,
484            } => {
485                state.groups.insert(group, stored);
486                if let Some(reverse) = new_reverse {
487                    state.user_groups.insert(owner, reverse);
488                } else {
489                    state
490                        .user_groups
491                        .get_mut(&owner)
492                        .expect("validated owner reverse index")
493                        .insert(group);
494                }
495            }
496            Mutation::User {
497                group,
498                user,
499                from,
500                to,
501                previous_owners,
502                new_reverse,
503                ..
504            } => {
505                let stored = state.groups.get_mut(&group).expect("validated group");
506                match to {
507                    Some(role) => {
508                        stored.users.insert(user, role);
509                    }
510                    None => {
511                        stored.users.remove(&user);
512                    }
513                }
514                stored.owners = owner_count_after(previous_owners, from, to);
515                stored.revision = callback_txid;
516                match (from, to) {
517                    (None, Some(_)) => {
518                        if let Some(reverse) = new_reverse {
519                            state.user_groups.insert(user, reverse);
520                        } else {
521                            state
522                                .user_groups
523                                .get_mut(&user)
524                                .expect("validated user reverse index")
525                                .insert(group);
526                        }
527                    }
528                    (Some(_), None) => {
529                        let reverse = state
530                            .user_groups
531                            .get_mut(&user)
532                            .expect("validated user reverse index");
533                        reverse.remove(&group);
534                        if reverse.is_empty() {
535                            state.user_groups.remove(&user);
536                        }
537                    }
538                    _ => {}
539                }
540            }
541            Mutation::Model {
542                group,
543                model,
544                from,
545                to,
546                new_reverse,
547                ..
548            } => {
549                let stored = state.groups.get_mut(&group).expect("validated group");
550                if to {
551                    stored.models.insert(model);
552                } else {
553                    stored.models.remove(&model);
554                }
555                stored.revision = callback_txid;
556                match (from, to) {
557                    (false, true) => {
558                        if let Some(reverse) = new_reverse {
559                            state.model_groups.insert(model, reverse);
560                        } else {
561                            state
562                                .model_groups
563                                .get_mut(&model)
564                                .expect("validated model reverse index")
565                                .insert(group);
566                        }
567                    }
568                    (true, false) => {
569                        let reverse = state
570                            .model_groups
571                            .get_mut(&model)
572                            .expect("validated model reverse index");
573                        reverse.remove(&group);
574                        if reverse.is_empty() {
575                            state.model_groups.remove(&model);
576                        }
577                    }
578                    _ => {}
579                }
580            }
581        }
582        state.cursor = Some(callback_txid);
583        Ok(prepared.result.into_public())
584    }
585
586    fn ensure_available(&self) -> Result<(), String> {
587        if self.unavailable.load(Ordering::SeqCst) {
588            Err("projection is unavailable until reopen".to_owned())
589        } else {
590            Ok(())
591        }
592    }
593
594    fn mark_unavailable(&self) {
595        self.unavailable.store(true, Ordering::SeqCst);
596    }
597}
598
599fn prepare_user_change(
600    state: &mut State,
601    group: GroupId,
602    actor: UserId,
603    user: UserId,
604    to: Option<GroupRole>,
605    callback_txid: TxId,
606) -> Result<Prepared, String> {
607    let Some(stored) = state.groups.get(&group) else {
608        return Ok(Prepared::rejected("group does not exist"));
609    };
610    let Some(actor_role) = stored.users.get(&actor).copied() else {
611        return Ok(Prepared::rejected("actor is not authorized"));
612    };
613    let from = stored.users.get(&user).copied();
614    let permitted = match actor_role {
615        GroupRole::User => false,
616        GroupRole::Admin => matches!(
617            (from, to),
618            (None, Some(GroupRole::User))
619                | (Some(GroupRole::User), None)
620                | (Some(GroupRole::User), Some(GroupRole::User))
621        ),
622        GroupRole::Owner => true,
623    };
624    if !permitted {
625        return Ok(Prepared::rejected(match actor_role {
626            GroupRole::Admin => "administrator transition is not permitted",
627            _ => "actor is not authorized",
628        }));
629    }
630    if actor_role == GroupRole::Owner
631        && from == Some(GroupRole::Owner)
632        && to != Some(GroupRole::Owner)
633        && stored.owners == 1
634    {
635        return Ok(Prepared::rejected(
636            "final owner cannot be removed or demoted",
637        ));
638    }
639    let previous_revision = stored.revision;
640    let previous_owners = stored.owners;
641    if from == to {
642        return Ok(Prepared::unchanged(group, previous_revision));
643    }
644    let mut new_reverse = None;
645    if from.is_none() && to.is_some() {
646        state
647            .groups
648            .get_mut(&group)
649            .expect("existing group")
650            .users
651            .try_reserve(1)
652            .map_err(|error| format!("reserve group user membership: {error}"))?;
653        if let Some(reverse) = state.user_groups.get_mut(&user) {
654            reverse
655                .try_reserve(1)
656                .map_err(|error| format!("reserve user reverse membership: {error}"))?;
657        } else {
658            state
659                .user_groups
660                .try_reserve(1)
661                .map_err(|error| format!("reserve user reverse index: {error}"))?;
662            let mut reverse = HashSet::new();
663            reverse
664                .try_reserve(1)
665                .map_err(|error| format!("reserve user reverse membership: {error}"))?;
666            reverse.insert(group);
667            new_reverse = Some(reverse);
668        }
669    }
670    Ok(Prepared {
671        result: PreparedResult::Applied(GroupRevision {
672            group_id: group,
673            txid: callback_txid,
674        }),
675        mutation: Mutation::User {
676            group,
677            user,
678            from,
679            to,
680            previous_revision,
681            previous_owners,
682            new_reverse,
683        },
684    })
685}
686
687fn prepare_model_change(
688    state: &mut State,
689    group: GroupId,
690    actor: UserId,
691    model: ModelId,
692    to: bool,
693    callback_txid: TxId,
694) -> Result<Prepared, String> {
695    let Some(stored) = state.groups.get(&group) else {
696        return Ok(Prepared::rejected("group does not exist"));
697    };
698    if stored.users.get(&actor) != Some(&GroupRole::Owner) {
699        return Ok(Prepared::rejected("actor is not authorized"));
700    }
701    let from = stored.models.contains(&model);
702    let previous_revision = stored.revision;
703    if from == to {
704        return Ok(Prepared::unchanged(group, previous_revision));
705    }
706    let mut new_reverse = None;
707    if to {
708        state
709            .groups
710            .get_mut(&group)
711            .expect("existing group")
712            .models
713            .try_reserve(1)
714            .map_err(|error| format!("reserve group model membership: {error}"))?;
715        if let Some(reverse) = state.model_groups.get_mut(&model) {
716            reverse
717                .try_reserve(1)
718                .map_err(|error| format!("reserve model reverse membership: {error}"))?;
719        } else {
720            state
721                .model_groups
722                .try_reserve(1)
723                .map_err(|error| format!("reserve model reverse index: {error}"))?;
724            let mut reverse = HashSet::new();
725            reverse
726                .try_reserve(1)
727                .map_err(|error| format!("reserve model reverse membership: {error}"))?;
728            reverse.insert(group);
729            new_reverse = Some(reverse);
730        }
731    }
732    Ok(Prepared {
733        result: PreparedResult::Applied(GroupRevision {
734            group_id: group,
735            txid: callback_txid,
736        }),
737        mutation: Mutation::Model {
738            group,
739            model,
740            from,
741            to,
742            previous_revision,
743            new_reverse,
744        },
745    })
746}
747
748struct Prepared {
749    result: PreparedResult,
750    mutation: Mutation,
751}
752
753impl Prepared {
754    fn rejected(reason: &str) -> Self {
755        Self {
756            result: PreparedResult::Rejected(reason.to_owned()),
757            mutation: Mutation::None,
758        }
759    }
760
761    fn unchanged(group_id: GroupId, txid: TxId) -> Self {
762        Self {
763            result: PreparedResult::Unchanged(GroupRevision { group_id, txid }),
764            mutation: Mutation::None,
765        }
766    }
767}
768
769enum PreparedResult {
770    Applied(GroupRevision),
771    Unchanged(GroupRevision),
772    Rejected(String),
773}
774
775impl PreparedResult {
776    fn into_public(self) -> ApplyOutcome {
777        match self {
778            Self::Applied(revision) => ApplyOutcome::Applied(revision),
779            Self::Unchanged(revision) => ApplyOutcome::Unchanged(revision),
780            Self::Rejected(reason) => ApplyOutcome::Rejected(reason),
781        }
782    }
783}
784
785enum Mutation {
786    None,
787    Create {
788        group: GroupId,
789        owner: UserId,
790        stored: GroupState,
791        new_reverse: Option<HashSet<GroupId>>,
792    },
793    User {
794        group: GroupId,
795        user: UserId,
796        from: Option<GroupRole>,
797        to: Option<GroupRole>,
798        previous_revision: TxId,
799        previous_owners: usize,
800        new_reverse: Option<HashSet<GroupId>>,
801    },
802    Model {
803        group: GroupId,
804        model: ModelId,
805        from: bool,
806        to: bool,
807        previous_revision: TxId,
808        new_reverse: Option<HashSet<GroupId>>,
809    },
810}
811
812fn mutation_matches(state: &State, mutation: &Mutation) -> bool {
813    match mutation {
814        Mutation::None => true,
815        Mutation::Create {
816            group,
817            owner,
818            new_reverse,
819            ..
820        } => {
821            if state.groups.contains_key(group) {
822                return false;
823            }
824            match new_reverse {
825                Some(reverse) => !state.user_groups.contains_key(owner) && reverse.contains(group),
826                None => state
827                    .user_groups
828                    .get(owner)
829                    .is_some_and(|groups| !groups.contains(group)),
830            }
831        }
832        Mutation::User {
833            group,
834            user,
835            from,
836            to,
837            previous_revision,
838            previous_owners,
839            new_reverse,
840        } => {
841            let Some(stored) = state.groups.get(group) else {
842                return false;
843            };
844            if stored.revision != *previous_revision
845                || stored.owners != *previous_owners
846                || stored.users.get(user).copied() != *from
847            {
848                return false;
849            }
850            match (from, to) {
851                (None, Some(_)) => match new_reverse {
852                    Some(reverse) => {
853                        !state.user_groups.contains_key(user) && reverse.contains(group)
854                    }
855                    None => state
856                        .user_groups
857                        .get(user)
858                        .is_some_and(|groups| !groups.contains(group)),
859                },
860                (Some(_), None) => state
861                    .user_groups
862                    .get(user)
863                    .is_some_and(|groups| groups.contains(group)),
864                _ => true,
865            }
866        }
867        Mutation::Model {
868            group,
869            model,
870            from,
871            to,
872            previous_revision,
873            new_reverse,
874        } => {
875            let Some(stored) = state.groups.get(group) else {
876                return false;
877            };
878            if stored.revision != *previous_revision || stored.models.contains(model) != *from {
879                return false;
880            }
881            match (from, to) {
882                (false, true) => match new_reverse {
883                    Some(reverse) => {
884                        !state.model_groups.contains_key(model) && reverse.contains(group)
885                    }
886                    None => state
887                        .model_groups
888                        .get(model)
889                        .is_some_and(|groups| !groups.contains(group)),
890                },
891                (true, false) => state
892                    .model_groups
893                    .get(model)
894                    .is_some_and(|groups| groups.contains(group)),
895                _ => true,
896            }
897        }
898    }
899}
900
901fn owner_count_after(previous: usize, from: Option<GroupRole>, to: Option<GroupRole>) -> usize {
902    match (from == Some(GroupRole::Owner), to == Some(GroupRole::Owner)) {
903        (false, true) => previous + 1,
904        (true, false) => previous - 1,
905        _ => previous,
906    }
907}
908
909fn persist_prepared(
910    connection: &mut Connection,
911    callback_txid: TxId,
912    prepared: &Prepared,
913) -> Result<(), String> {
914    let transaction = connection
915        .transaction_with_behavior(TransactionBehavior::Immediate)
916        .map_err(|error| format!("begin projection apply transaction: {error}"))?;
917    match &prepared.mutation {
918        Mutation::None => {}
919        Mutation::Create {
920            group,
921            owner,
922            stored,
923            ..
924        } => {
925            expect_one(
926                transaction.execute(
927                    "INSERT INTO groups (group_id, revision) VALUES (?1, ?2)",
928                    params![
929                        group.txid().as_bytes().as_slice(),
930                        stored.revision.as_bytes().as_slice()
931                    ],
932                ),
933                "insert group",
934            )?;
935            let owner_txid = owner.as_tx_id();
936            expect_one(
937                transaction.execute(
938                    "INSERT INTO user_memberships (group_id, user_id, role) VALUES (?1, ?2, ?3)",
939                    params![
940                        group.txid().as_bytes().as_slice(),
941                        owner_txid.as_bytes().as_slice(),
942                        GroupRole::Owner.stored()
943                    ],
944                ),
945                "insert owner membership",
946            )?;
947        }
948        Mutation::User {
949            group,
950            user,
951            from,
952            to,
953            ..
954        } => {
955            let user_txid = user.as_tx_id();
956            match (from, to) {
957                (None, Some(role)) => expect_one(
958                    transaction.execute(
959                        "INSERT INTO user_memberships (group_id, user_id, role) VALUES (?1, ?2, ?3)",
960                        params![
961                            group.txid().as_bytes().as_slice(),
962                            user_txid.as_bytes().as_slice(),
963                            role.stored()
964                        ],
965                    ),
966                    "insert user membership",
967                )?,
968                (Some(_), Some(role)) => expect_one(
969                    transaction.execute(
970                        "UPDATE user_memberships SET role = ?3 WHERE group_id = ?1 AND user_id = ?2",
971                        params![
972                            group.txid().as_bytes().as_slice(),
973                            user_txid.as_bytes().as_slice(),
974                            role.stored()
975                        ],
976                    ),
977                    "update user membership",
978                )?,
979                (Some(_), None) => expect_one(
980                    transaction.execute(
981                        "DELETE FROM user_memberships WHERE group_id = ?1 AND user_id = ?2",
982                        params![
983                            group.txid().as_bytes().as_slice(),
984                            user_txid.as_bytes().as_slice()
985                        ],
986                    ),
987                    "delete user membership",
988                )?,
989                (None, None) => {
990                    return Err("prepared user mutation has no state change".to_owned());
991                }
992            }
993            update_group_revision(&transaction, *group, callback_txid)?;
994        }
995        Mutation::Model {
996            group,
997            model,
998            from,
999            to,
1000            ..
1001        } => {
1002            match (from, to) {
1003                (false, true) => expect_one(
1004                    transaction.execute(
1005                        "INSERT INTO model_memberships (group_id, model_id) VALUES (?1, ?2)",
1006                        params![
1007                            group.txid().as_bytes().as_slice(),
1008                            model.as_bytes().as_slice()
1009                        ],
1010                    ),
1011                    "insert model membership",
1012                )?,
1013                (true, false) => expect_one(
1014                    transaction.execute(
1015                        "DELETE FROM model_memberships WHERE group_id = ?1 AND model_id = ?2",
1016                        params![
1017                            group.txid().as_bytes().as_slice(),
1018                            model.as_bytes().as_slice()
1019                        ],
1020                    ),
1021                    "delete model membership",
1022                )?,
1023                _ => return Err("prepared model mutation has no state change".to_owned()),
1024            }
1025            update_group_revision(&transaction, *group, callback_txid)?;
1026        }
1027    }
1028    expect_one(
1029        transaction.execute(
1030            "UPDATE metadata SET last_applied_txid = ?1 WHERE singleton = 1",
1031            params![callback_txid.as_bytes().as_slice()],
1032        ),
1033        "advance projection cursor",
1034    )?;
1035    transaction
1036        .commit()
1037        .map_err(|error| format!("commit projection apply transaction: {error}"))
1038}
1039
1040fn update_group_revision(
1041    transaction: &rusqlite::Transaction<'_>,
1042    group: GroupId,
1043    callback_txid: TxId,
1044) -> Result<(), String> {
1045    expect_one(
1046        transaction.execute(
1047            "UPDATE groups SET revision = ?2 WHERE group_id = ?1",
1048            params![
1049                group.txid().as_bytes().as_slice(),
1050                callback_txid.as_bytes().as_slice()
1051            ],
1052        ),
1053        "update group revision",
1054    )
1055}
1056
1057fn expect_one(result: Result<usize, SqlError>, operation: &str) -> Result<(), String> {
1058    let changed = result.map_err(|error| format!("{operation}: {error}"))?;
1059    if changed == 1 {
1060        Ok(())
1061    } else {
1062        Err(format!("{operation}: expected one row, changed {changed}"))
1063    }
1064}
1065
1066fn persist_clear(connection: &mut Connection) -> Result<(), String> {
1067    let transaction = connection
1068        .transaction_with_behavior(TransactionBehavior::Immediate)
1069        .map_err(|error| format!("begin projection clear transaction: {error}"))?;
1070    transaction
1071        .execute("DELETE FROM user_memberships", [])
1072        .map_err(|error| format!("clear user memberships: {error}"))?;
1073    transaction
1074        .execute("DELETE FROM model_memberships", [])
1075        .map_err(|error| format!("clear model memberships: {error}"))?;
1076    transaction
1077        .execute("DELETE FROM groups", [])
1078        .map_err(|error| format!("clear groups: {error}"))?;
1079    expect_one(
1080        transaction.execute(
1081            "UPDATE metadata SET last_applied_txid = NULL WHERE singleton = 1",
1082            [],
1083        ),
1084        "clear projection cursor",
1085    )?;
1086    transaction
1087        .commit()
1088        .map_err(|error| format!("commit projection clear transaction: {error}"))
1089}
1090
1091fn copy_set<K>(set: Option<&HashSet<K>>, operation: &str) -> Result<Vec<K>, String>
1092where
1093    K: Copy + Eq + Hash,
1094{
1095    let Some(set) = set else {
1096        return Ok(Vec::new());
1097    };
1098    let mut values = Vec::new();
1099    values
1100        .try_reserve(set.len())
1101        .map_err(|error| format!("reserve {operation}: {error}"))?;
1102    values.extend(set.iter().copied());
1103    Ok(values)
1104}
1105
1106enum LoadError {
1107    Recoverable,
1108    Fatal(String),
1109}
1110
1111fn load_or_rebuild(
1112    database: &Path,
1113    ordering: &K1TxnOrdering,
1114) -> Result<(Connection, State), String> {
1115    let exists = database.try_exists().map_err(|error| {
1116        format!(
1117            "inspect projection database {}: {error}",
1118            database.display()
1119        )
1120    })?;
1121    if !exists {
1122        return initialize_database(database);
1123    }
1124    match load_existing(database, ordering) {
1125        Ok(loaded) => Ok(loaded),
1126        Err(LoadError::Recoverable) => {
1127            remove_database_files(database)?;
1128            initialize_database(database)
1129        }
1130        Err(LoadError::Fatal(error)) => Err(error),
1131    }
1132}
1133
1134fn initialize_database(database: &Path) -> Result<(Connection, State), String> {
1135    let mut connection = Connection::open(database)
1136        .map_err(|error| format!("create projection database {}: {error}", database.display()))?;
1137    let journal: String = connection
1138        .query_row("PRAGMA journal_mode = WAL", [], |row| row.get(0))
1139        .map_err(|error| format!("enable projection WAL: {error}"))?;
1140    if !journal.eq_ignore_ascii_case("wal") {
1141        return Err(format!("enable projection WAL: SQLite selected {journal}"));
1142    }
1143    connection
1144        .execute_batch("PRAGMA synchronous = FULL; PRAGMA foreign_keys = ON;")
1145        .map_err(|error| format!("configure projection database: {error}"))?;
1146    let transaction = connection
1147        .transaction_with_behavior(TransactionBehavior::Immediate)
1148        .map_err(|error| format!("begin projection initialization: {error}"))?;
1149    for (statement, operation) in [
1150        (CREATE_METADATA, "create metadata schema"),
1151        (CREATE_GROUPS, "create groups schema"),
1152        (CREATE_USERS, "create user membership schema"),
1153        (CREATE_MODELS, "create model membership schema"),
1154        (CREATE_USER_INDEX, "create user reverse index"),
1155        (CREATE_MODEL_INDEX, "create model reverse index"),
1156    ] {
1157        transaction
1158            .execute(statement, [])
1159            .map_err(|error| format!("{operation}: {error}"))?;
1160    }
1161    expect_one(
1162        transaction.execute(
1163            "INSERT INTO metadata (singleton, schema_version, last_applied_txid) VALUES (1, 1, NULL)",
1164            [],
1165        ),
1166        "initialize projection metadata",
1167    )?;
1168    transaction
1169        .commit()
1170        .map_err(|error| format!("commit projection initialization: {error}"))?;
1171    Ok((connection, State::default()))
1172}
1173
1174fn load_existing(
1175    database: &Path,
1176    ordering: &K1TxnOrdering,
1177) -> Result<(Connection, State), LoadError> {
1178    let connection = Connection::open(database).map_err(|error| {
1179        classify_validation_error(
1180            error,
1181            &format!("open projection database {}", database.display()),
1182        )
1183    })?;
1184    let journal: String = validate_sql(
1185        connection.query_row("PRAGMA journal_mode", [], |row| row.get(0)),
1186        "read projection journal mode",
1187    )?;
1188    if !journal.eq_ignore_ascii_case("wal") {
1189        return Err(LoadError::Recoverable);
1190    }
1191    validate_sql(
1192        connection.execute_batch("PRAGMA synchronous = FULL; PRAGMA foreign_keys = ON;"),
1193        "configure projection database",
1194    )?;
1195    validate_quick_check(&connection)?;
1196    validate_schema(&connection)?;
1197    validate_foreign_keys(&connection)?;
1198    let state = load_state(&connection, ordering)?;
1199    Ok((connection, state))
1200}
1201
1202fn validate_quick_check(connection: &Connection) -> Result<(), LoadError> {
1203    let mut statement = validate_sql(
1204        connection.prepare("PRAGMA quick_check"),
1205        "prepare projection quick_check",
1206    )?;
1207    let mut rows = validate_sql(statement.query([]), "run projection quick_check")?;
1208    let Some(row) = validate_sql(rows.next(), "read projection quick_check")? else {
1209        return Err(LoadError::Recoverable);
1210    };
1211    let result: String = validate_sql(row.get(0), "decode projection quick_check")?;
1212    if result != "ok" || validate_sql(rows.next(), "finish projection quick_check")?.is_some() {
1213        return Err(LoadError::Recoverable);
1214    }
1215    Ok(())
1216}
1217
1218#[derive(Eq, Ord, PartialEq, PartialOrd)]
1219struct SchemaObject {
1220    name: String,
1221    kind: String,
1222    table: String,
1223    sql: Option<String>,
1224}
1225
1226fn validate_schema(connection: &Connection) -> Result<(), LoadError> {
1227    let mut statement = validate_sql(
1228        connection.prepare(
1229            "SELECT name, type, tbl_name, sql FROM sqlite_master ORDER BY name, type, tbl_name",
1230        ),
1231        "prepare projection schema validation",
1232    )?;
1233    let mut rows = validate_sql(statement.query([]), "query projection schema")?;
1234    let mut actual = Vec::new();
1235    while let Some(row) = validate_sql(rows.next(), "read projection schema")? {
1236        actual.push(SchemaObject {
1237            name: validate_sql(row.get(0), "decode projection schema name")?,
1238            kind: validate_sql(row.get(1), "decode projection schema type")?,
1239            table: validate_sql(row.get(2), "decode projection schema table")?,
1240            sql: validate_sql(row.get(3), "decode projection schema SQL")?,
1241        });
1242    }
1243    let mut expected = vec![
1244        schema("groups", "table", "groups", Some(CREATE_GROUPS)),
1245        schema("metadata", "table", "metadata", Some(CREATE_METADATA)),
1246        schema(
1247            "model_memberships",
1248            "table",
1249            "model_memberships",
1250            Some(CREATE_MODELS),
1251        ),
1252        schema(
1253            "model_memberships_by_model",
1254            "index",
1255            "model_memberships",
1256            Some(CREATE_MODEL_INDEX),
1257        ),
1258        schema("sqlite_autoindex_groups_1", "index", "groups", None),
1259        schema(
1260            "sqlite_autoindex_model_memberships_1",
1261            "index",
1262            "model_memberships",
1263            None,
1264        ),
1265        schema(
1266            "sqlite_autoindex_user_memberships_1",
1267            "index",
1268            "user_memberships",
1269            None,
1270        ),
1271        schema(
1272            "user_memberships",
1273            "table",
1274            "user_memberships",
1275            Some(CREATE_USERS),
1276        ),
1277        schema(
1278            "user_memberships_by_user",
1279            "index",
1280            "user_memberships",
1281            Some(CREATE_USER_INDEX),
1282        ),
1283    ];
1284    expected.sort();
1285    if actual != expected {
1286        return Err(LoadError::Recoverable);
1287    }
1288    Ok(())
1289}
1290
1291fn schema(name: &str, kind: &str, table: &str, sql: Option<&str>) -> SchemaObject {
1292    SchemaObject {
1293        name: name.to_owned(),
1294        kind: kind.to_owned(),
1295        table: table.to_owned(),
1296        sql: sql.map(str::to_owned),
1297    }
1298}
1299
1300fn validate_foreign_keys(connection: &Connection) -> Result<(), LoadError> {
1301    let mut statement = validate_sql(
1302        connection.prepare("PRAGMA foreign_key_check"),
1303        "prepare projection foreign key validation",
1304    )?;
1305    let mut rows = validate_sql(statement.query([]), "run projection foreign key validation")?;
1306    if validate_sql(rows.next(), "read projection foreign key validation")?.is_some() {
1307        return Err(LoadError::Recoverable);
1308    }
1309    Ok(())
1310}
1311
1312fn load_state(connection: &Connection, ordering: &K1TxnOrdering) -> Result<State, LoadError> {
1313    let mut state = State::default();
1314    let mut metadata = validate_sql(
1315        connection.prepare(
1316            "SELECT singleton, schema_version, last_applied_txid FROM metadata ORDER BY singleton",
1317        ),
1318        "prepare projection metadata",
1319    )?;
1320    let mut metadata_rows = validate_sql(metadata.query([]), "query projection metadata")?;
1321    let Some(row) = validate_sql(metadata_rows.next(), "read projection metadata")? else {
1322        return Err(LoadError::Recoverable);
1323    };
1324    let singleton: i64 = validate_sql(row.get(0), "decode metadata singleton")?;
1325    let version: i64 = validate_sql(row.get(1), "decode metadata version")?;
1326    let cursor: Option<Vec<u8>> = validate_sql(row.get(2), "decode metadata cursor")?;
1327    if singleton != 1
1328        || version != 1
1329        || validate_sql(metadata_rows.next(), "finish projection metadata")?.is_some()
1330    {
1331        return Err(LoadError::Recoverable);
1332    }
1333    state.cursor = match cursor {
1334        Some(bytes) => Some(TxId::from_bytes(exact_bytes::<12>(&bytes)?)),
1335        None => None,
1336    };
1337
1338    let group_count = table_count(connection, "groups")?;
1339    state
1340        .groups
1341        .try_reserve(group_count)
1342        .map_err(|error| LoadError::Fatal(format!("reserve group state: {error}")))?;
1343    let mut groups = validate_sql(
1344        connection.prepare("SELECT group_id, revision FROM groups"),
1345        "prepare stored groups",
1346    )?;
1347    let mut group_rows = validate_sql(groups.query([]), "query stored groups")?;
1348    while let Some(row) = validate_sql(group_rows.next(), "read stored group")? {
1349        let group_bytes: Vec<u8> = validate_sql(row.get(0), "decode stored group ID")?;
1350        let revision_bytes: Vec<u8> = validate_sql(row.get(1), "decode stored group revision")?;
1351        let group = GroupId::new(TxId::from_bytes(exact_bytes::<12>(&group_bytes)?));
1352        let revision = TxId::from_bytes(exact_bytes::<12>(&revision_bytes)?);
1353        if state
1354            .groups
1355            .insert(
1356                group,
1357                GroupState {
1358                    revision,
1359                    users: HashMap::new(),
1360                    models: HashSet::new(),
1361                    owners: 0,
1362                },
1363            )
1364            .is_some()
1365        {
1366            return Err(LoadError::Recoverable);
1367        }
1368    }
1369
1370    let mut users = validate_sql(
1371        connection.prepare("SELECT group_id, user_id, role FROM user_memberships"),
1372        "prepare stored user memberships",
1373    )?;
1374    let mut user_rows = validate_sql(users.query([]), "query stored user memberships")?;
1375    while let Some(row) = validate_sql(user_rows.next(), "read stored user membership")? {
1376        let group_bytes: Vec<u8> = validate_sql(row.get(0), "decode membership group ID")?;
1377        let user_bytes: Vec<u8> = validate_sql(row.get(1), "decode membership user ID")?;
1378        let role_value: i64 = validate_sql(row.get(2), "decode membership role")?;
1379        let group = GroupId::new(TxId::from_bytes(exact_bytes::<12>(&group_bytes)?));
1380        let user = UserId::from_tx_id(TxId::from_bytes(exact_bytes::<12>(&user_bytes)?));
1381        let Some(role) = GroupRole::from_stored(role_value) else {
1382            return Err(LoadError::Recoverable);
1383        };
1384        let Some(stored) = state.groups.get_mut(&group) else {
1385            return Err(LoadError::Recoverable);
1386        };
1387        stored
1388            .users
1389            .try_reserve(1)
1390            .map_err(|error| LoadError::Fatal(format!("reserve stored users: {error}")))?;
1391        if stored.users.insert(user, role).is_some() {
1392            return Err(LoadError::Recoverable);
1393        }
1394        if role == GroupRole::Owner {
1395            stored.owners = stored
1396                .owners
1397                .checked_add(1)
1398                .ok_or_else(|| LoadError::Fatal("stored owner count overflow".to_owned()))?;
1399        }
1400        insert_reverse(
1401            &mut state.user_groups,
1402            user,
1403            group,
1404            "stored user reverse index",
1405        )?;
1406    }
1407
1408    let mut models = validate_sql(
1409        connection.prepare("SELECT group_id, model_id FROM model_memberships"),
1410        "prepare stored model memberships",
1411    )?;
1412    let mut model_rows = validate_sql(models.query([]), "query stored model memberships")?;
1413    while let Some(row) = validate_sql(model_rows.next(), "read stored model membership")? {
1414        let group_bytes: Vec<u8> = validate_sql(row.get(0), "decode model membership group ID")?;
1415        let model_bytes: Vec<u8> = validate_sql(row.get(1), "decode model membership model ID")?;
1416        let group = GroupId::new(TxId::from_bytes(exact_bytes::<12>(&group_bytes)?));
1417        let model = ModelId::from_bytes(exact_bytes::<32>(&model_bytes)?);
1418        let Some(stored) = state.groups.get_mut(&group) else {
1419            return Err(LoadError::Recoverable);
1420        };
1421        stored
1422            .models
1423            .try_reserve(1)
1424            .map_err(|error| LoadError::Fatal(format!("reserve stored models: {error}")))?;
1425        if !stored.models.insert(model) {
1426            return Err(LoadError::Recoverable);
1427        }
1428        insert_reverse(
1429            &mut state.model_groups,
1430            model,
1431            group,
1432            "stored model reverse index",
1433        )?;
1434    }
1435
1436    if state.groups.values().any(|group| group.owners == 0)
1437        || (!state.groups.is_empty() && state.cursor.is_none())
1438    {
1439        return Err(LoadError::Recoverable);
1440    }
1441    validate_canonical_ids(&state, ordering)?;
1442    Ok(state)
1443}
1444
1445fn table_count(connection: &Connection, table: &str) -> Result<usize, LoadError> {
1446    let sql = format!("SELECT COUNT(*) FROM {table}");
1447    let count: i64 = validate_sql(
1448        connection.query_row(&sql, [], |row| row.get(0)),
1449        "count projection rows",
1450    )?;
1451    usize::try_from(count).map_err(|_| LoadError::Recoverable)
1452}
1453
1454fn insert_reverse<K>(
1455    reverse: &mut HashMap<K, HashSet<GroupId>>,
1456    key: K,
1457    group: GroupId,
1458    operation: &str,
1459) -> Result<(), LoadError>
1460where
1461    K: Copy + Eq + Hash,
1462{
1463    if !reverse.contains_key(&key) {
1464        reverse
1465            .try_reserve(1)
1466            .map_err(|error| LoadError::Fatal(format!("reserve {operation}: {error}")))?;
1467    }
1468    let groups = reverse.entry(key).or_default();
1469    groups
1470        .try_reserve(1)
1471        .map_err(|error| LoadError::Fatal(format!("reserve {operation}: {error}")))?;
1472    if !groups.insert(group) {
1473        return Err(LoadError::Recoverable);
1474    }
1475    Ok(())
1476}
1477
1478fn validate_canonical_ids(state: &State, ordering: &K1TxnOrdering) -> Result<(), LoadError> {
1479    let subsystem = SubsystemId::from_str(SUBSYSTEM)
1480        .map_err(|error| LoadError::Fatal(format!("construct groups subsystem ID: {error}")))?;
1481    let mut validated = HashSet::new();
1482    validated
1483        .try_reserve(
1484            state
1485                .groups
1486                .len()
1487                .checked_mul(2)
1488                .and_then(|count| count.checked_add(1))
1489                .ok_or_else(|| LoadError::Fatal("canonical ID count overflow".to_owned()))?,
1490        )
1491        .map_err(|error| LoadError::Fatal(format!("reserve canonical ID validation: {error}")))?;
1492    for txid in state
1493        .groups
1494        .iter()
1495        .flat_map(|(group, stored)| [group.txid(), stored.revision])
1496        .chain(state.cursor)
1497    {
1498        if validated.contains(&txid) {
1499            continue;
1500        }
1501        let bytes = ordering.get_txn(txid).map_err(|error| {
1502            LoadError::Fatal(format!("validate canonical transaction {txid}: {error}"))
1503        })?;
1504        let Some(bytes) = bytes else {
1505            return Err(LoadError::Recoverable);
1506        };
1507        let transaction = Transaction::parse(&bytes).map_err(|_| LoadError::Recoverable)?;
1508        if transaction.subsystem() != subsystem {
1509            return Err(LoadError::Recoverable);
1510        }
1511        validated.insert(txid);
1512    }
1513    Ok(())
1514}
1515
1516fn exact_bytes<const N: usize>(bytes: &[u8]) -> Result<[u8; N], LoadError> {
1517    bytes.try_into().map_err(|_| LoadError::Recoverable)
1518}
1519
1520fn validate_sql<T>(result: Result<T, SqlError>, operation: &str) -> Result<T, LoadError> {
1521    result.map_err(|error| classify_validation_error(error, operation))
1522}
1523
1524fn classify_validation_error(error: SqlError, operation: &str) -> LoadError {
1525    let recoverable = match &error {
1526        SqlError::SqliteFailure(code, _)
1527            if matches!(
1528                code.code,
1529                rusqlite::ffi::ErrorCode::DatabaseCorrupt | rusqlite::ffi::ErrorCode::NotADatabase
1530            ) =>
1531        {
1532            true
1533        }
1534        SqlError::FromSqlConversionFailure(..)
1535        | SqlError::IntegralValueOutOfRange(..)
1536        | SqlError::Utf8Error(..)
1537        | SqlError::InvalidColumnType(..)
1538        | SqlError::QueryReturnedNoRows => true,
1539        _ => false,
1540    };
1541    if recoverable {
1542        LoadError::Recoverable
1543    } else {
1544        LoadError::Fatal(format!("{operation}: {error}"))
1545    }
1546}
1547
1548fn remove_database_files(database: &Path) -> Result<(), String> {
1549    for path in [
1550        sidecar_path(database, "-wal"),
1551        sidecar_path(database, "-shm"),
1552        database.to_path_buf(),
1553    ] {
1554        match fs::remove_file(&path) {
1555            Ok(()) => {}
1556            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
1557            Err(error) => {
1558                return Err(format!(
1559                    "remove recoverable projection file {}: {error}",
1560                    path.display()
1561                ));
1562            }
1563        }
1564    }
1565    Ok(())
1566}
1567
1568fn sidecar_path(database: &Path, suffix: &str) -> PathBuf {
1569    let mut path = database.as_os_str().to_os_string();
1570    path.push(suffix);
1571    PathBuf::from(path)
1572}
1573
1574#[cfg(test)]
1575mod internal_tests {
1576    use std::{
1577        sync::{Arc, TryLockError, mpsc},
1578        thread,
1579        time::{Duration, Instant},
1580    };
1581
1582    use kcode_k1_txn_ordering::K1TxnOrdering;
1583    use rusqlite::Connection;
1584    use tempfile::TempDir;
1585
1586    use super::{GroupAction, GroupId, Projection, TxId, UserId};
1587
1588    #[test]
1589    fn blocked_apply_does_not_block_queries() {
1590        let temporary = TempDir::new().unwrap();
1591        let ordering = K1TxnOrdering::open(&temporary.path().join("ordering")).unwrap();
1592        let projection_root = temporary.path().join("projection");
1593        let (projection, _) = Projection::open(&projection_root, &ordering).unwrap();
1594        let projection = Arc::new(projection);
1595        let blocker = Connection::open(projection_root.join("groups.sqlite3")).unwrap();
1596        blocker.execute_batch("BEGIN IMMEDIATE").unwrap();
1597        let applying = Arc::clone(&projection);
1598        let (apply_sent, apply_received) = mpsc::channel();
1599        let apply = thread::spawn(move || {
1600            apply_sent
1601                .send(applying.apply(
1602                    TxId::from_bytes([1; 12]),
1603                    GroupAction::Create {
1604                        owner: UserId::from_tx_id(TxId::from_bytes([2; 12])),
1605                    },
1606                ))
1607                .unwrap();
1608        });
1609        let waiting_started = Instant::now();
1610        loop {
1611            match projection.apply_lane.try_lock() {
1612                Err(TryLockError::WouldBlock) => break,
1613                Err(TryLockError::Poisoned(_)) => panic!("apply lane poisoned"),
1614                Ok(lane) => drop(lane),
1615            }
1616            assert!(waiting_started.elapsed() < Duration::from_secs(1));
1617            thread::yield_now();
1618        }
1619        assert!(matches!(
1620            apply_received.try_recv(),
1621            Err(mpsc::TryRecvError::Empty)
1622        ));
1623        let querying = Arc::clone(&projection);
1624        let (query_sent, query_received) = mpsc::channel();
1625        let query = thread::spawn(move || {
1626            query_sent
1627                .send(querying.get(GroupId::new(TxId::from_bytes([1; 12]))))
1628                .unwrap();
1629        });
1630        let query_result = query_received.recv_timeout(Duration::from_secs(1));
1631        blocker.execute_batch("ROLLBACK").unwrap();
1632        assert!(query_result.unwrap().unwrap().is_none());
1633        assert!(
1634            apply_received
1635                .recv_timeout(Duration::from_secs(1))
1636                .unwrap()
1637                .is_ok()
1638        );
1639        query.join().unwrap();
1640        apply.join().unwrap();
1641    }
1642}
kcode_k1_groups_projection/lib.rs

kcode_k1_groups_projection/
lib.rs
