slices.network
Auto-indexing service and GraphQL API for AT Protocol Records
quickslice.slices.network/
atproto
gleam
graphql
chadtmiller.com
1import gleam/erlang/process.{type Subject}
2import gleam/list
3import group_registry
4
5/// Represents the operation type for a record event
6pub type RecordOperation {
7 Create
8 Update
9 Delete
10}
11
12/// Event published when a record is created, updated, or deleted
13pub type RecordEvent {
14 RecordEvent(
15 uri: String,
16 cid: String,
17 did: String,
18 collection: String,
19 value: String,
20 indexed_at: String,
21 operation: RecordOperation,
22 )
23}
24
25/// The group name for all record event subscriptions
26const group_name = "record_events"
27
28// Global registry name - must be created once and reused
29@external(erlang, "pubsub_ffi", "get_registry_name")
30fn registry_name() -> process.Name(group_registry.Message(RecordEvent))
31
32/// Initialize the PubSub registry
33/// Must be called once when the server starts
34pub fn start() -> Nil {
35 // Start the registry (idempotent - safe to call multiple times)
36 let _ = group_registry.start(registry_name())
37 Nil
38}
39
40fn get_registry() -> group_registry.GroupRegistry(RecordEvent) {
41 // Get existing registry
42 group_registry.get_registry(registry_name())
43}
44
45/// Subscribe to all record events
46/// Returns a Subject that will receive RecordEvent messages
47pub fn subscribe() -> Subject(RecordEvent) {
48 let registry = get_registry()
49 let my_pid = process.self()
50 group_registry.join(registry, group_name, my_pid)
51}
52
53/// Unsubscribe from record events
54pub fn unsubscribe(subscriber: Subject(RecordEvent)) -> Nil {
55 let registry = get_registry()
56 case process.subject_owner(subscriber) {
57 Ok(my_pid) -> group_registry.leave(registry, group_name, [my_pid])
58 Error(_) -> Nil
59 }
60}
61
62/// Publish a record event to all subscribers
63pub fn publish(event: RecordEvent) -> Nil {
64 let registry = get_registry()
65 let subscribers = group_registry.members(registry, group_name)
66 list.each(subscribers, fn(sub) { process.send(sub, event) })
67}