daemon/
auto_rooms.rs

1//! Daemon-side reactive auto-room maintenance.
2//!
3//! Every `Session` SET fires this saga, which ensures that the
4//! anchor-rooms derived from the session's identity exist and that
5//! the session is a member of them. The four anchor kinds:
6//!
7//! - `everyone`              — singleton, every session always a member.
8//! - `host:<name>`           — when `Session.host.name` is populated.
9//! - `op:<operator>`         — when `Session.operator` is populated.
10//! - `project:<basename>`    — when `Session.project` is populated
11//!   (the shim resolves it from `git rev-parse --show-toplevel`).
12//!
13//! Idempotent: re-emitting a Room SET that's identical to the existing
14//! row is a no-op for the registry, and the composite-id RoomMember
15//! rows survive a re-SET unchanged. So the saga can fire freely on
16//! every Session SET without producing churn in the event log.
17//!
18//! The saga runs as a `DispatchAutoRooms` server-internal command (the
19//! pattern mirrors `DedupeNicknameSaga` → `DedupeNicknames`).
20
21use std::sync::Arc;
22
23use chrono::Utc;
24use marshal_entities::{
25    AutoSource, GetAllRoomMembers, GetAllRooms, Room, RoomId, RoomKind, RoomMember, RoomMemberId,
26    Session,
27};
28use myko::{
29    command::{CommandContext, CommandError, CommandHandler},
30    myko_command,
31    prelude::myko_saga,
32    saga::{SagaContext, SagaHandler},
33    wire::{MEvent, MEventType},
34};
35
36/// Force-link the saga registrations from this module against
37/// dead-code elimination.
38pub fn link() {}
39
40// ─── Saga ───────────────────────────────────────────────────────────────────
41
42#[myko_saga]
43pub struct AutoRoomSaga;
44
45impl SagaHandler for AutoRoomSaga {
46    type EventItem = Session;
47    type Command = DispatchAutoRooms;
48    const EVENT_TYPE: MEventType = MEventType::SET;
49
50    fn handle(session: Session, _event: MEvent, _ctx: Arc<SagaContext>) -> Option<Self::Command> {
51        Some(DispatchAutoRooms {
52            session_id: session.id.0.as_ref().to_string(),
53        })
54    }
55}
56
57// ─── Command ────────────────────────────────────────────────────────────────
58
59/// Server-internal command that ensures the four auto-rooms anchored
60/// on a session's identity exist, and that the session is a member of
61/// each. Pure idempotent reconciliation: missing room → SET; missing
62/// membership → SET; nothing missing → no-op.
63#[myko_command]
64pub struct DispatchAutoRooms {
65    pub session_id: String,
66}
67
68impl CommandHandler for DispatchAutoRooms {
69    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
70        // Find the session row this saga was triggered by. If it's
71        // gone (DEL'd between the SET and our run), bail — the saga
72        // is responsible for live identity, not stale state.
73        let sessions: Vec<Arc<Session>> = ctx.exec_query(marshal_entities::GetAllSessions {})?;
74        let Some(session) = sessions
75            .iter()
76            .find(|s| s.id.0.as_ref() == self.session_id.as_str())
77            .cloned()
78        else {
79            return Ok(());
80        };
81
82        let rooms: Vec<Arc<Room>> = ctx.exec_query(GetAllRooms {})?;
83        let memberships: Vec<Arc<RoomMember>> = ctx.exec_query(GetAllRoomMembers {})?;
84        let now = Utc::now().timestamp_millis();
85
86        // Compute the desired anchor-rooms for this session.
87        let mut anchors: Vec<(RoomId, String, AutoSource)> = vec![(
88            RoomId(Arc::from("everyone")),
89            "everyone".to_string(),
90            AutoSource::Everyone,
91        )];
92        if let Some(host) = session.host.as_ref()
93            && !host.name.is_empty()
94        {
95            let id = format!("host:{}", host.name);
96            anchors.push((
97                RoomId(Arc::from(id.as_str())),
98                id,
99                AutoSource::Host {
100                    name: host.name.clone(),
101                },
102            ));
103        }
104        if let Some(op) = session.operator.as_ref()
105            && !op.is_empty()
106        {
107            let id = format!("op:{op}");
108            anchors.push((
109                RoomId(Arc::from(id.as_str())),
110                id,
111                AutoSource::Operator { name: op.clone() },
112            ));
113        }
114        if let Some(project) = session.project.as_ref()
115            && !project.is_empty()
116        {
117            let id = format!("project:{project}");
118            anchors.push((
119                RoomId(Arc::from(id.as_str())),
120                id,
121                AutoSource::Project {
122                    basename: project.clone(),
123                },
124            ));
125        }
126
127        for (room_id, name, source) in anchors {
128            // Ensure the Room entity exists (idempotent SET).
129            let already_room = rooms.iter().any(|r| r.id == room_id);
130            if !already_room {
131                ctx.emit_set(&Room {
132                    id: room_id.clone(),
133                    name,
134                    description: None,
135                    kind: RoomKind::Auto { source },
136                    created_at: now,
137                })?;
138            }
139
140            // Ensure the membership row exists.
141            let member_id = RoomMember::make_id(room_id.0.as_ref(), session.id.0.as_ref());
142            let already_member = memberships
143                .iter()
144                .any(|m| m.id.0.as_ref() == member_id.as_str());
145            if !already_member {
146                ctx.emit_set(&RoomMember {
147                    id: RoomMemberId(Arc::from(member_id.as_str())),
148                    room_id: room_id.clone(),
149                    session_id: session.id.clone(),
150                    joined_at: now,
151                })?;
152            }
153        }
154
155        Ok(())
156    }
157}
daemon/auto_rooms.rs

daemon/
auto_rooms.rs
