nx_event_queue.nx source

1// nx_event_queue.nx -- substrate-level event dispatch primitive. 2// 3// FIFO event queue with kind-based dispatching. Many primitives I've 4// shipped emit events (xenocell intrusion, treaty breach, health 5// transition, scam pattern). nx_event_queue is the generic substrate 6// for accumulating + dispatching those events to subscribers. 7// 8// Composes: 9// nx_xenocell -- intrusion observations flow into events 10// nx_homeostasis -- migration signals flow into events 11// nx_health_check -- RED transitions flow into events 12// nx_book -- queue-drained events get journaled 13// 14// V1 ships a single global queue per caller; multiplexing across 15// subscribers is the caller's job (V1 returns the next event by 16// kind filter). 17 18import "nx_syscalls.nx" 19import "nx_tier.nx" 20 21// ===== Sealed enum: NxEventKind ==================================== 22// 23// Generic event kinds the substrate emits. Caller-extensible: 24// values >= NX_EV_N_KINDS may be defined by callers for app-specific 25// events; substrate does not validate them. 26 27const NX_EV_NONE: nx_int = 0 28const NX_EV_XENO_OBSERVATION: nx_int = 1 29const NX_EV_HOMEOSTASIS_SIGNAL: nx_int = 2 30const NX_EV_HEALTH_TRANSITION: nx_int = 3 31const NX_EV_TREATY_BREACHED: nx_int = 4 32const NX_EV_SCAM_FLAGGED: nx_int = 5 33const NX_EV_RECOVERY_TRIGGERED: nx_int = 6 34const NX_EV_TWIN_KEY_DECISION: nx_int = 7 35const NX_EV_VESICLE_DELIVERED: nx_int = 8 36const NX_EV_N_KINDS: nx_int = 9 37 38const NX_EQ_OK: nx_int = 0 39const NX_EQ_ERR_FULL: nx_int = 1 40const NX_EQ_ERR_EMPTY: nx_int = 2 41const NX_EQ_ERR_BAD_KIND: nx_int = 3 42 43// ===== Struct: NxEvent ============================================= 44 45struct NxEvent { 46 ts_us: nx_size, 47 kind: nx_int, 48 subject_id: nx_int, // who/what the event concerns 49 payload: nx_size, // arbitrary payload value 50 priority: nx_int, // 0=low, 100=normal, 200=high 51} 52 53// ===== Struct: NxEventQueue ======================================== 54 55struct NxEventQueue { 56 events: *NxEvent, 57 capacity: nx_size, 58 head: nx_size, 59 tail: nx_size, 60 count: nx_size, 61 total_emitted: nx_int, 62} 63 64const NX_EV_BYTES: nx_size = 40 65 66func nx_ev_kind_is_known(k: nx_int) -> nx_int { 67 if k < 0 { return 0 } 68 if k >= NX_EV_N_KINDS { return 0 } 69 return 1 70} 71 72func nx_event_queue_new(capacity: nx_size) -> *NxEventQueue { 73 let q: *NxEventQueue = (sys_mmap(48)) as *NxEventQueue 74 let bytes: nx_size = capacity * NX_EV_BYTES 75 q.events = (sys_mmap(bytes)) as *NxEvent 76 q.capacity = capacity 77 q.head = 0 78 q.tail = 0 79 q.count = 0 80 q.total_emitted = 0 81 return q 82} 83 84func _ev_at(q: *NxEventQueue, idx: nx_size) -> *NxEvent { 85 return (q.events as i64 + (idx as i64) * NX_EV_BYTES) as *NxEvent 86} 87 88// ===== nx_event_queue_emit ======================================== 89 90func nx_event_queue_emit(q: *NxEventQueue, 91 ts_us: nx_size, 92 kind: nx_int, 93 subject_id: nx_int, 94 payload: nx_size, 95 priority: nx_int) -> nx_int { 96 if q.count >= q.capacity { return NX_EQ_ERR_FULL } 97 let e: *NxEvent = _ev_at(q, q.head) 98 e.ts_us = ts_us 99 e.kind = kind 100 e.subject_id = subject_id 101 e.payload = payload 102 e.priority = priority 103 q.head = q.head + 1 104 if q.head >= q.capacity { q.head = 0 } 105 q.count = q.count + 1 106 q.total_emitted = q.total_emitted + 1 107 return NX_EQ_OK 108} 109 110// ===== nx_event_queue_drain_one =================================== 111// 112// Pop oldest event into caller-supplied out_kind/out_subject. Returns 113// NX_EQ_OK or NX_EQ_ERR_EMPTY. 114 115func nx_event_queue_drain_one(q: *NxEventQueue, 116 out_kind: *i64, 117 out_subject: *i64, 118 out_payload: *i64, 119 out_ts_us: *i64) -> nx_int { 120 if q.count == 0 { return NX_EQ_ERR_EMPTY } 121 let e: *NxEvent = _ev_at(q, q.tail) 122 out_kind[0] = e.kind as i64 123 out_subject[0] = e.subject_id as i64 124 out_payload[0] = e.payload as i64 125 out_ts_us[0] = e.ts_us as i64 126 q.tail = q.tail + 1 127 if q.tail >= q.capacity { q.tail = 0 } 128 q.count = q.count - 1 129 return NX_EQ_OK 130} 131 132// ===== nx_event_queue_peek_kind =================================== 133// 134// Return the kind of the next event without removing it. -1 if empty. 135 136func nx_event_queue_peek_kind(q: *NxEventQueue) -> nx_int { 137 if q.count == 0 { return -1 } 138 let e: *NxEvent = _ev_at(q, q.tail) 139 return e.kind 140} 141 142func nx_event_queue_count(q: *NxEventQueue) -> nx_size { 143 return q.count 144} 145 146func nx_event_queue_total_emitted(q: *NxEventQueue) -> nx_int { 147 return q.total_emitted 148} 149 150func nx_event_queue_count_by_kind(q: *NxEventQueue, kind: nx_int) -> nx_int { 151 var hits: nx_int = 0 152 var live: nx_size = q.count 153 if live > q.capacity { live = q.capacity } 154 var i: nx_size = 0 155 var idx: nx_size = q.tail 156 while i < live { 157 let e: *NxEvent = _ev_at(q, idx) 158 if e.kind == kind { hits = hits + 1 } 159 idx = idx + 1 160 if idx >= q.capacity { idx = 0 } 161 i = i + 1 162 } 163 return hits 164}