1use std::{
2 collections::HashMap,
3 path::Path,
4 sync::{Arc, Mutex},
5};
6
7use kcode_k1_groups_projection::{ApplyOutcome, GroupAction, Projection};
8use kcode_k1_peering::K1Peering;
9use kcode_k1_transaction::SubsystemId;
10use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem as K1Subsystem};
11
12pub use kcode_k1_groups_projection::{
13 ALL_MODELS, ALL_USERS, Group, GroupId, GroupMemberships, GroupRevision, GroupRole, GroupUser,
14 LOCAL_MODELS, ModelId, SentinelGroup, TxId, UserId,
15};
16
17const SUBSYSTEM_NAME: &str = "k1-groups-subsystem";
18const CREATE_LENGTH: usize = 30;
19const SET_USER_ROLE_LENGTH: usize = 55;
20const SET_MODEL_MEMBERSHIP_LENGTH: usize = 75;
21
22type OperationId = [u8; 16];
23
24#[derive(Clone, Debug, Eq, PartialEq)]
25enum Action {
26 Create {
27 owner: UserId,
28 },
29 SetUserRole {
30 group: GroupId,
31 actor: UserId,
32 user: UserId,
33 role: Option<GroupRole>,
34 },
35 SetModelMembership {
36 group: GroupId,
37 actor: UserId,
38 model: ModelId,
39 present: bool,
40 },
41}
42
43impl Action {
44 fn projection_action(&self) -> GroupAction {
45 match self {
46 Self::Create { owner } => GroupAction::Create { owner: *owner },
47 Self::SetUserRole {
48 group,
49 actor,
50 user,
51 role,
52 } => GroupAction::SetUserRole {
53 group: *group,
54 actor: *actor,
55 user: *user,
56 role: *role,
57 },
58 Self::SetModelMembership {
59 group,
60 actor,
61 model,
62 present,
63 } => GroupAction::SetModelMembership {
64 group: *group,
65 actor: *actor,
66 model: *model,
67 present: *present,
68 },
69 }
70 }
71
72 fn payload(&self, operation_id: OperationId) -> Vec<u8> {
73 let capacity = match self {
74 Self::Create { .. } => CREATE_LENGTH,
75 Self::SetUserRole { .. } => SET_USER_ROLE_LENGTH,
76 Self::SetModelMembership { .. } => SET_MODEL_MEMBERSHIP_LENGTH,
77 };
78 let mut payload = Vec::with_capacity(capacity);
79 payload.extend_from_slice(&[1, self.kind()]);
80 payload.extend_from_slice(&operation_id);
81 match self {
82 Self::Create { owner } => payload.extend_from_slice(owner.as_tx_id().as_bytes()),
83 Self::SetUserRole {
84 group,
85 actor,
86 user,
87 role,
88 } => {
89 payload.extend_from_slice(group.txid().as_bytes());
90 payload.extend_from_slice(actor.as_tx_id().as_bytes());
91 payload.extend_from_slice(user.as_tx_id().as_bytes());
92 payload.push(role_byte(*role));
93 }
94 Self::SetModelMembership {
95 group,
96 actor,
97 model,
98 present,
99 } => {
100 payload.extend_from_slice(group.txid().as_bytes());
101 payload.extend_from_slice(actor.as_tx_id().as_bytes());
102 payload.extend_from_slice(model.as_bytes());
103 payload.push(u8::from(*present));
104 }
105 }
106 payload
107 }
108
109 fn kind(&self) -> u8 {
110 match self {
111 Self::Create { .. } => 1,
112 Self::SetUserRole { .. } => 2,
113 Self::SetModelMembership { .. } => 3,
114 }
115 }
116}
117
118#[derive(Clone)]
119enum Resolution {
120 Applied(GroupRevision),
121 Unchanged(GroupRevision),
122 Rejected(String),
123}
124
125impl Resolution {
126 fn kind(&self) -> OutcomeKind {
127 match self {
128 Self::Applied(_) => OutcomeKind::Applied,
129 Self::Unchanged(_) => OutcomeKind::Unchanged,
130 Self::Rejected(_) => OutcomeKind::Rejected,
131 }
132 }
133
134 fn result(self) -> Result<GroupRevision, String> {
135 match self {
136 Self::Applied(revision) | Self::Unchanged(revision) => Ok(revision),
137 Self::Rejected(error) => Err(error),
138 }
139 }
140}
141
142impl From<ApplyOutcome> for Resolution {
143 fn from(outcome: ApplyOutcome) -> Self {
144 match outcome {
145 ApplyOutcome::Applied(revision) => Self::Applied(revision),
146 ApplyOutcome::Unchanged(revision) => Self::Unchanged(revision),
147 ApplyOutcome::Rejected(error) => Self::Rejected(error),
148 }
149 }
150}
151
152#[derive(Clone)]
153struct CallbackRecord {
154 txid: TxId,
155 resolution: Resolution,
156}
157
158enum CallbackEvidence {
159 None,
160 One(CallbackRecord),
161 Ambiguous,
162}
163
164struct PendingOperation {
165 action: Action,
166 evidence: CallbackEvidence,
167}
168
169struct FacadeState {
170 available: bool,
171 pending: HashMap<OperationId, PendingOperation>,
172}
173
174struct SharedState {
175 inner: Mutex<FacadeState>,
176}
177
178impl SharedState {
179 fn new() -> Self {
180 Self {
181 inner: Mutex::new(FacadeState {
182 available: true,
183 pending: HashMap::new(),
184 }),
185 }
186 }
187
188 fn ensure_available(&self) -> Result<(), String> {
189 let state = self
190 .inner
191 .lock()
192 .map_err(|_| "k1 groups state lock failed".to_string())?;
193 if state.available {
194 Ok(())
195 } else {
196 Err("k1 groups instance unavailable".to_string())
197 }
198 }
199
200 fn reserve(&self, action: Action) -> Result<OperationId, String> {
201 self.ensure_available()?;
202 loop {
203 let mut operation_id = [0_u8; 16];
204 getrandom::fill(&mut operation_id)
205 .map_err(|error| format!("operation ID randomness failed: {error}"))?;
206 let mut state = self
207 .inner
208 .lock()
209 .map_err(|_| "k1 groups state lock failed".to_string())?;
210 if !state.available {
211 return Err("k1 groups instance unavailable".to_string());
212 }
213 if state.pending.contains_key(&operation_id) {
214 continue;
215 }
216 state
217 .pending
218 .try_reserve(1)
219 .map_err(|_| "pending operation allocation failed".to_string())?;
220 state.pending.insert(
221 operation_id,
222 PendingOperation {
223 action,
224 evidence: CallbackEvidence::None,
225 },
226 );
227 return Ok(operation_id);
228 }
229 }
230
231 fn record(
232 &self,
233 operation_id: OperationId,
234 action: &Action,
235 txid: TxId,
236 resolution: Resolution,
237 ) -> Result<(), String> {
238 let mut state = self
239 .inner
240 .lock()
241 .map_err(|_| "k1 groups state lock failed".to_string())?;
242 if !state.available {
243 return Err("k1 groups instance unavailable".to_string());
244 }
245 let contradiction = match state.pending.get_mut(&operation_id) {
246 None => return Ok(()),
247 Some(pending) if &pending.action != action => {
248 Some("operation ID collided with a different action")
249 }
250 Some(pending) => match pending.evidence {
251 CallbackEvidence::None => {
252 pending.evidence = CallbackEvidence::One(CallbackRecord { txid, resolution });
253 None
254 }
255 CallbackEvidence::One(_) | CallbackEvidence::Ambiguous => {
256 Some("duplicate callback evidence for one operation ID")
257 }
258 },
259 };
260 if let Some(error) = contradiction {
261 make_unavailable(&mut state);
262 Err(error.to_string())
263 } else {
264 Ok(())
265 }
266 }
267
268 fn finish(
269 &self,
270 operation_id: OperationId,
271 submission: Result<TxId, String>,
272 ) -> Result<GroupRevision, String> {
273 let evidence = {
274 let mut state = self
275 .inner
276 .lock()
277 .map_err(|_| "k1 groups state lock failed".to_string())?;
278 let pending = state.pending.remove(&operation_id);
279 if !state.available {
280 CallbackEvidence::Ambiguous
281 } else {
282 pending
283 .map(|pending| pending.evidence)
284 .unwrap_or(CallbackEvidence::Ambiguous)
285 }
286 };
287 let summary = evidence.summary();
288 match reconcile_decision(&submission, summary) {
289 ReconcileDecision::UseCallback(expected_kind) => match evidence {
290 CallbackEvidence::One(record) if record.resolution.kind() == expected_kind => {
291 record.resolution.result()
292 }
293 _ => self.fail_resolution("callback reconciliation state contradiction"),
294 },
295 ReconcileDecision::UseSubmissionError => match submission {
296 Err(error) => Err(error),
297 Ok(_) => self.fail_resolution("submission reconciliation state contradiction"),
298 },
299 ReconcileDecision::Fault(error) => self.fail_resolution(error),
300 }
301 }
302
303 fn fail_resolution<T>(&self, error: &str) -> Result<T, String> {
304 let _ = self.make_unavailable();
305 Err(error.to_string())
306 }
307
308 fn make_unavailable(&self) -> Result<(), String> {
309 let mut state = self
310 .inner
311 .lock()
312 .map_err(|_| "k1 groups state lock failed".to_string())?;
313 make_unavailable(&mut state);
314 Ok(())
315 }
316}
317
318fn make_unavailable(state: &mut FacadeState) {
319 state.available = false;
320 for pending in state.pending.values_mut() {
321 pending.evidence = CallbackEvidence::Ambiguous;
322 }
323}
324
325#[derive(Clone, Copy, Debug, Eq, PartialEq)]
326enum OutcomeKind {
327 Applied,
328 Unchanged,
329 Rejected,
330}
331
332#[derive(Clone, Copy, Debug, Eq, PartialEq)]
333enum EvidenceSummary {
334 None,
335 One { txid: TxId, kind: OutcomeKind },
336 Ambiguous,
337}
338
339impl CallbackEvidence {
340 fn summary(&self) -> EvidenceSummary {
341 match self {
342 Self::None => EvidenceSummary::None,
343 Self::One(record) => EvidenceSummary::One {
344 txid: record.txid,
345 kind: record.resolution.kind(),
346 },
347 Self::Ambiguous => EvidenceSummary::Ambiguous,
348 }
349 }
350}
351
352#[derive(Clone, Copy, Debug, Eq, PartialEq)]
353enum ReconcileDecision {
354 UseCallback(OutcomeKind),
355 UseSubmissionError,
356 Fault(&'static str),
357}
358
359fn reconcile_decision(
360 submission: &Result<TxId, String>,
361 evidence: EvidenceSummary,
362) -> ReconcileDecision {
363 match evidence {
364 EvidenceSummary::Ambiguous => {
365 ReconcileDecision::Fault("ambiguous callback evidence made instance unavailable")
366 }
367 EvidenceSummary::None => {
368 if submission.is_ok() {
369 ReconcileDecision::Fault("successful submission had no synchronous callback")
370 } else {
371 ReconcileDecision::UseSubmissionError
372 }
373 }
374 EvidenceSummary::One { txid, kind } => match submission {
375 Ok(returned) if *returned == txid => ReconcileDecision::UseCallback(kind),
376 Ok(_) => ReconcileDecision::Fault(
377 "submission transaction ID did not match callback transaction ID",
378 ),
379 Err(_) => ReconcileDecision::UseCallback(kind),
380 },
381 }
382}
383
384struct GroupsSubsystem {
385 projection: Arc<Projection>,
386 shared: Arc<SharedState>,
387}
388
389impl K1Subsystem for GroupsSubsystem {
390 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
391 let (operation_id, action) = match parse_payload(payload) {
392 Ok(parsed) => parsed,
393 Err(error) => {
394 let _ = self.shared.make_unavailable();
395 return Err(error);
396 }
397 };
398 self.shared.ensure_available()?;
399 let outcome = match self.projection.apply(id, action.projection_action()) {
400 Ok(outcome) => outcome,
401 Err(error) => {
402 let _ = self.shared.make_unavailable();
403 return Err(error);
404 }
405 };
406 self.shared
407 .record(operation_id, &action, id, outcome.into())
408 }
409
410 fn reorg(&self) -> Result<(), String> {
411 let state_result = self.shared.make_unavailable();
412 match self.projection.clear() {
413 Ok(()) => state_result,
414 Err(error) => Err(error),
415 }
416 }
417}
418
419pub struct K1Groups {
420 projection: Arc<Projection>,
421 _ordering: Arc<K1TxnOrdering>,
422 peering: Arc<K1Peering>,
423 shared: Arc<SharedState>,
424 subsystem: SubsystemId,
425}
426
427impl K1Groups {
428 pub fn open(
429 root: &Path,
430 ordering: Arc<K1TxnOrdering>,
431 peering: Arc<K1Peering>,
432 ) -> Result<Self, String> {
433 let (projection, checkpoint) = Projection::open(root, &ordering)?;
434 let projection = Arc::new(projection);
435 let shared = Arc::new(SharedState::new());
436 let subsystem = SubsystemId::from_str(SUBSYSTEM_NAME)?;
437 let callback = Arc::new(GroupsSubsystem {
438 projection: projection.clone(),
439 shared: shared.clone(),
440 });
441 if let Err(error) = ordering.register_subsystem(subsystem, checkpoint, callback) {
442 let _ = shared.make_unavailable();
443 return Err(error);
444 }
445 Ok(Self {
446 projection,
447 _ordering: ordering,
448 peering,
449 shared,
450 subsystem,
451 })
452 }
453
454 pub fn create(&self, owner: UserId) -> Result<GroupRevision, String> {
455 self.mutate(Action::Create { owner })
456 }
457
458 pub fn set_user_role(
459 &self,
460 actor: UserId,
461 group: GroupId,
462 user: UserId,
463 role: Option<GroupRole>,
464 ) -> Result<GroupRevision, String> {
465 self.mutate(Action::SetUserRole {
466 group,
467 actor,
468 user,
469 role,
470 })
471 }
472
473 pub fn set_model_membership(
474 &self,
475 actor: UserId,
476 group: GroupId,
477 model: ModelId,
478 present: bool,
479 ) -> Result<GroupRevision, String> {
480 self.mutate(Action::SetModelMembership {
481 group,
482 actor,
483 model,
484 present,
485 })
486 }
487
488 pub fn get(&self, group: GroupId) -> Result<Option<Group>, String> {
489 self.query(|projection| projection.get(group))
490 }
491
492 pub fn groups_for_user(&self, user: UserId) -> Result<Vec<GroupId>, String> {
493 self.query(|projection| projection.groups_for_user(user))
494 }
495
496 pub fn groups_for_model(&self, model: ModelId) -> Result<Vec<GroupId>, String> {
497 self.query(|projection| projection.groups_for_model(model))
498 }
499
500 pub fn memberships(&self, user: UserId, model: ModelId) -> Result<GroupMemberships, String> {
501 self.query(|projection| projection.memberships(user, model))
502 }
503
504 fn mutate(&self, action: Action) -> Result<GroupRevision, String> {
505 let operation_id = self.shared.reserve(action.clone())?;
506 let payload = action.payload(operation_id);
507 let submission = self.peering.submit_txn(self.subsystem, &payload);
508 self.shared.finish(operation_id, submission)
509 }
510
511 fn query<T>(&self, query: impl FnOnce(&Projection) -> Result<T, String>) -> Result<T, String> {
512 self.shared.ensure_available()?;
513 let result = match query(&self.projection) {
514 Ok(result) => result,
515 Err(error) => {
516 let _ = self.shared.make_unavailable();
517 return Err(error);
518 }
519 };
520 self.shared.ensure_available()?;
521 Ok(result)
522 }
523}
524
525fn parse_payload(payload: &[u8]) -> Result<(OperationId, Action), String> {
526 if payload.len() < 2 {
527 return Err("malformed groups payload header".to_string());
528 }
529 if payload[0] != 1 {
530 return Err("unknown groups payload version".to_string());
531 }
532 match payload[1] {
533 1 => parse_create(payload),
534 2 => parse_set_user_role(payload),
535 3 => parse_set_model_membership(payload),
536 _ => Err("unknown groups payload kind".to_string()),
537 }
538}
539
540fn parse_create(payload: &[u8]) -> Result<(OperationId, Action), String> {
541 require_length(payload, CREATE_LENGTH)?;
542 Ok((
543 operation_id(payload),
544 Action::Create {
545 owner: UserId::from_tx_id(txid(payload, 18)),
546 },
547 ))
548}
549
550fn parse_set_user_role(payload: &[u8]) -> Result<(OperationId, Action), String> {
551 require_length(payload, SET_USER_ROLE_LENGTH)?;
552 let role = match payload[54] {
553 0 => None,
554 1 => Some(GroupRole::User),
555 2 => Some(GroupRole::Admin),
556 3 => Some(GroupRole::Owner),
557 _ => return Err("invalid group role discriminant".to_string()),
558 };
559 Ok((
560 operation_id(payload),
561 Action::SetUserRole {
562 group: GroupId::new(txid(payload, 18)),
563 actor: UserId::from_tx_id(txid(payload, 30)),
564 user: UserId::from_tx_id(txid(payload, 42)),
565 role,
566 },
567 ))
568}
569
570fn parse_set_model_membership(payload: &[u8]) -> Result<(OperationId, Action), String> {
571 require_length(payload, SET_MODEL_MEMBERSHIP_LENGTH)?;
572 let present = match payload[74] {
573 0 => false,
574 1 => true,
575 _ => return Err("invalid model presence discriminant".to_string()),
576 };
577 let mut model = [0_u8; 32];
578 model.copy_from_slice(&payload[42..74]);
579 Ok((
580 operation_id(payload),
581 Action::SetModelMembership {
582 group: GroupId::new(txid(payload, 18)),
583 actor: UserId::from_tx_id(txid(payload, 30)),
584 model: ModelId::from_bytes(model),
585 present,
586 },
587 ))
588}
589
590fn require_length(payload: &[u8], expected: usize) -> Result<(), String> {
591 if payload.len() == expected {
592 Ok(())
593 } else {
594 Err(format!(
595 "wrong groups payload length: expected {expected}, received {}",
596 payload.len()
597 ))
598 }
599}
600
601fn operation_id(payload: &[u8]) -> OperationId {
602 let mut operation_id = [0_u8; 16];
603 operation_id.copy_from_slice(&payload[2..18]);
604 operation_id
605}
606
607fn txid(payload: &[u8], start: usize) -> TxId {
608 let mut bytes = [0_u8; 12];
609 bytes.copy_from_slice(&payload[start..start + 12]);
610 TxId::from_bytes(bytes)
611}
612
613fn role_byte(role: Option<GroupRole>) -> u8 {
614 match role {
615 None => 0,
616 Some(GroupRole::User) => 1,
617 Some(GroupRole::Admin) => 2,
618 Some(GroupRole::Owner) => 3,
619 }
620}
621
622#[cfg(test)]
623mod tests;