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