neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

Keryx: dotted, id-keyed topics a rabbit user can subscribe to

8a95d37 · Gabriel Morell · 2026-10-05 22:41

24 files +119 -33

Message

{commit_body(@commit)}

Files changed

modified apps/room_hermes/lib/room_hermes/accounts.ex

+11 −0

@@ -114,4 +114,15 @@ defmodule RoomHermes.Accounts do

114	114	from(r in RabbitUser, where: r.username == ^username and r.password == ^password)
115	115	\|> Repo.one()
116	116	end
	117	+
	118	+ @doc """
	119	+ Whether an account owns a Keryx.
	120	+
	121	+ Schemaless, because the Keryx is the sanctum's: this app shares its
	122	+ database but not its schemas.
	123	+ """
	124	+ def owns_keryx?(user_id, keryx_id) do
	125	+ from(k in "keryxiae", where: k.id == ^keryx_id and k.user_id == ^user_id)
	126	+ \|> Repo.exists?()
	127	+ end
117	128	end

modified apps/room_hermes/lib/room_hermes_web/controllers/topic_controller.ex

+33 −2

@@ -10,7 +10,8 @@ defmodule RoomHermesWeb.TopicController do

10	10	Which routing keys a rabbit user may read and write.
11	11
12	12	Reading is the whole of its own topic: the board, the image beside it, and
13		− anything else a Pythiae publishes there.
	13	+ anything else a Pythiae publishes there. A rabbit user may also read every
	14	+ Keryx its account owns -- see `keryx_permitted?/3`.
14	15
15	16	Writing is confined to an uplink prefix, `<topic>.up.`. A client is not
16	17	allowed to write the topic a Pythiae publishes on, because everything else

@@ -31,7 +32,8 @@ defmodule RoomHermesWeb.TopicController do

31	32	send_resp(conn, :ok, "deny")
32	33
33	34	user ->
34		− if permitted?(user.topic, permission, routing_key) do
	35	+ if permitted?(user.topic, permission, routing_key) or
	36	+ keryx_permitted?(user, permission, routing_key) do
35	37	send_resp(conn, :ok, "allow")
36	38	else
37	39	send_resp(conn, :ok, "deny")

@@ -79,4 +81,33 @@ defmodule RoomHermesWeb.TopicController do

79	81
80	82	@doc "The channel a client asks for something on."
81	83	def request_prefix(topic), do: topic <> ".publish."
	84	+
	85	+ @doc """
	86	+ Whether a rabbit user may touch a Keryx's routing key -- the keys
	87	+ `RoomSanctum.Configuration.Keryx.topic/1` names.
	88	+
	89	+ Its account has to own the Keryx. Then it may read anything under it --
	90	+ a wildcard included, which is the point -- and write only to its control
	91	+ key, which is how a client asks for a refresh.
	92	+
	93	+ mix run -e 'IO.inspect RoomHermesWeb.TopicController.keryx_permission?("write", "control")'
	94	+ """
	95	+ def keryx_permitted?(%RabbitUser{user_id: uid}, permission, routing_key)
	96	+ when not is_nil(uid) do
	97	+ case Regex.run(~r/^apollos\.keryx\.(\d+)\.(.+)$/, routing_key) do
	98	+ [_, keryx_id, rest] ->
	99	+ keryx_permission?(permission, rest) and
	100	+ Accounts.owns_keryx?(uid, String.to_integer(keryx_id))
	101	+
	102	+ _ ->
	103	+ false
	104	+ end
	105	+ end
	106	+
	107	+ def keryx_permitted?(_user, _permission, _routing_key), do: false
	108	+
	109	+ @doc "The part of the Keryx rule that needs no database: what, under a Keryx, may be done."
	110	+ def keryx_permission?("read", _rest), do: true
	111	+ def keryx_permission?("write", "control"), do: true
	112	+ def keryx_permission?(_permission, _rest), do: false
82	113	end

modified apps/room_sanctum/lib/room_sanctum/configuration/keryx.ex

+24 −0

@@ -19,4 +19,28 @@ defmodule RoomSanctum.Configuration.Keryx do

19	19	\|> validate_required([:name, :query_ids, :ttl])
20	20	\|> unique_constraint(:name)
21	21	end
	22	+
	23	+ @doc """
	24	+ Where a Keryx publishes, as an AMQP routing key.
	25	+
	26	+ Dotted, because the topic exchange splits on dots and nothing else: a key
	27	+ of `/apollos/keryx/<name>/gtfs/12` is one word, so nobody could subscribe to
	28	+ a Keryx without naming every key in it. Keyed by id rather than name, since
	29	+ a name may have a dot in it and an id never does.
	30	+
	31	+ apollos.keryx.<id>.<type>.<query_id> the query's results
	32	+ apollos.keryx.<id>.<type>.<query_id>.meta the condensed summary
	33	+ apollos.keryx.<id>.<type>.<query_id>.<item>... single items
	34	+ apollos.keryx.<id>.control commands, inbound
	35	+
	36	+ An MQTT client sees the same thing with slashes -- see `mqtt_topic/1`.
	37	+ Who may read and write them is `RoomHermesWeb.TopicController`'s to say.
	38	+ """
	39	+ def topic(keryx_id), do: "apollos.keryx.#{keryx_id}"
	40	+
	41	+ @doc "Where a Keryx listens for commands."
	42	+ def control_topic(keryx_id), do: topic(keryx_id) <> ".control"
	43	+
	44	+ @doc "A routing key as an MQTT client spells it: RabbitMQ swaps `.` for `/`."
	45	+ def mqtt_topic(routing_key), do: String.replace(routing_key, ".", "/")
22	46	end

modified apps/room_sanctum/lib/room_sanctum/workers/keryx_worker.ex

+21 −15

@@ -5,6 +5,7 @@ defmodule RoomSanctum.Worker.Keryx do

5	5	require Logger
6	6
7	7	alias RoomSanctum.Configuration
	8	+ alias RoomSanctum.Configuration.Keryx
8	9
9	10	@registry :zeus
10	11

@@ -92,7 +93,7 @@ defmodule RoomSanctum.Worker.Keryx do

92	93
93	94	def handle_info(:subscribe_control, state) do
94	95	if state.keryx do
95		− control_topic = "/apollos/keryx/#{state.keryx.name}/control"
	96	+ control_topic = Keryx.control_topic(state.keryx.id)
96	97	queue_name = "keryx_control_#{state.keryx.id}"
97	98
98	99	case AMQP.Application.get_channel(:default) do

@@ -131,16 +132,18 @@ defmodule RoomSanctum.Worker.Keryx do

131	132	def handle_info({:basic_deliver, payload, _meta}, state) do
132	133	# Handle incoming control messages
133	134	case Poison.decode(payload) do
134		− {:ok, %{"action" => "refresh"}} ->
135		− Logger.info("Received refresh command for Keryx #{state.id}")
136		− query_workers(state.id)
137		− {:noreply, state}
138		−
	135	+ # The narrower match first: a map pattern matches any map with at least
	136	+ # those keys, so the bare refresh would otherwise swallow this one.
139	137	{:ok, %{"action" => "refresh", "query_id" => query_id}} ->
140	138	Logger.info("Received refresh command for Query #{query_id}")
141	139	refresh_single_query(state, query_id)
142	140	{:noreply, state}
143	141
	142	+ {:ok, %{"action" => "refresh"}} ->
	143	+ Logger.info("Received refresh command for Keryx #{state.id}")
	144	+ query_workers(state.id)
	145	+ {:noreply, state}
	146	+
144	147	_ ->
145	148	Logger.warn("Received unknown control message: #{payload}")
146	149	{:noreply, state}

@@ -311,8 +314,8 @@ defmodule RoomSanctum.Worker.Keryx do

311	314	\|> Enum.into(%{})
312	315	end
313	316
314		− defp publish_query_data(keryx_id, name, type, query_id, data, query) when is_list(data) do
315		− base_topic = "/apollos/keryx/#{name}/#{type}/#{query_id}"
	317	+ defp publish_query_data(keryx_id, _name, type, query_id, data, query) when is_list(data) do
	318	+ base_topic = query_topic(keryx_id, type, query_id)
316	319
317	320	# Cache items for the UI
318	321	RoomSanctum.Worker.KeryxItemCache.add_items(keryx_id, query_id, type, data)

@@ -333,7 +336,7 @@ defmodule RoomSanctum.Worker.Keryx do

333	336	data
334	337	\|> Enum.each(fn item ->
335	338	item_id = extract_item_id(item, type)
336		− item_topic = "#{base_topic}/#{item_id}"
	339	+ item_topic = "#{base_topic}.#{item_id}"
337	340	AMQP.Basic.publish(chan, "amq.topic", item_topic, item \|> Poison.encode!())
338	341	end)
339	342	end

@@ -343,9 +346,9 @@ defmodule RoomSanctum.Worker.Keryx do

343	346	end
344	347	end
345	348
346		− defp publish_query_data(_keryx_id, name, type, query_id, data, _query) do
	349	+ defp publish_query_data(keryx_id, _name, type, query_id, data, _query) do
347	350	# Handle non-list data (publish as-is, no caching)
348		− topic = "/apollos/keryx/#{name}/#{type}/#{query_id}"
	351	+ topic = query_topic(keryx_id, type, query_id)
349	352
350	353	case AMQP.Application.get_channel(:default) do
351	354	{:ok, chan} ->

@@ -356,12 +359,15 @@ defmodule RoomSanctum.Worker.Keryx do

356	359	end
357	360	end
358	361
	362	+ defp query_topic(keryx_id, type, query_id),
	363	+ do: "#{Keryx.topic(keryx_id)}.#{type}.#{query_id}"
	364	+
359	365	# Publish metadata about the query results using BasicMQTT condense
360	366	defp publish_metadata(chan, base_topic, type, query_id, data, query) do
361	367	# Use the existing condense function to get metadata
362	368	metadata = RoomSanctum.Condenser.BasicMQTT.condense({query_id, type}, data, query)
363	369
364		− meta_topic = "#{base_topic}/meta"
	370	+ meta_topic = "#{base_topic}.meta"
365	371	AMQP.Basic.publish(chan, "amq.topic", meta_topic, metadata \|> Poison.encode!())
366	372	end
367	373

@@ -372,7 +378,7 @@ defmodule RoomSanctum.Worker.Keryx do

372	378	\|> Enum.group_by(&extract_gtfs_route/1)
373	379	\|> Enum.each(fn {route, route_items} ->
374	380	# Publish all items for this route to <base>/<route>
375		− route_topic = "#{base_topic}/#{route}"
	381	+ route_topic = "#{base_topic}.#{route}"
376	382	AMQP.Basic.publish(chan, "amq.topic", route_topic, route_items \|> Poison.encode!())
377	383
378	384	# Group by destination within this route

@@ -380,14 +386,14 @@ defmodule RoomSanctum.Worker.Keryx do

380	386	\|> Enum.group_by(&extract_gtfs_destination/1)
381	387	\|> Enum.each(fn {destination, dest_items} ->
382	388	# Publish items for this route+destination to <base>/<route>/<destination>
383		− dest_topic = "#{base_topic}/#{route}/#{destination}"
	389	+ dest_topic = "#{base_topic}.#{route}.#{destination}"
384	390	AMQP.Basic.publish(chan, "amq.topic", dest_topic, dest_items \|> Poison.encode!())
385	391
386	392	# Also publish individual items
387	393	dest_items
388	394	\|> Enum.each(fn item ->
389	395	item_id = extract_item_id(item, :gtfs)
390		− item_topic = "#{base_topic}/#{route}/#{destination}/#{item_id}"
	396	+ item_topic = "#{base_topic}.#{route}.#{destination}.#{item_id}"
391	397	AMQP.Basic.publish(chan, "amq.topic", item_topic, item \|> Poison.encode!())
392	398	end)
393	399	end)

modified apps/room_sanctum/lib/room_sanctum_web/live/keryx_live/show.ex

+7 −0

@@ -2,6 +2,7 @@ defmodule RoomSanctumWeb.KeryxLive.Show do

2	2	use RoomSanctumWeb, :live_view_a
3	3
4	4	alias RoomSanctum.Configuration
	5	+ alias RoomSanctum.Configuration.Keryx
5	6
6	7	@impl true
7	8	def mount(_params, _session, socket) do

@@ -22,6 +23,7 @@ defmodule RoomSanctumWeb.KeryxLive.Show do

22	23	socket
23	24	\|> assign(:page_title, page_title(socket.assigns.live_action))
24	25	\|> assign(:keryx, keryx)
	26	+ \|> assign(:mqtt_base, mqtt_base(keryx))
25	27	\|> assign(:queries, queries)
26	28	\|> assign(:cached_items, cached_items)}
27	29	end

@@ -108,4 +110,9 @@ defmodule RoomSanctumWeb.KeryxLive.Show do

108	110	end)
109	111	end
110	112	end
	113	+
	114	+ # What a client subscribes to, which is not what the worker publishes to:
	115	+ # RabbitMQ shows an MQTT client the routing key with its dots as slashes.
	116	+ defp mqtt_base(keryx),
	117	+ do: keryx.id \|> Keryx.topic() \|> Keryx.mqtt_topic()
111	118	end

modified apps/room_sanctum/lib/room_sanctum_web/live/keryx_live/show.html.heex

+14 −14

@@ -47,19 +47,19 @@

47	47	<h3 class="card-title">Control Topic</h3>
48	48	<p class="text-sm opacity-75 mb-2">Publish commands to this topic to control the Keryx worker</p>
49	49	<div class="flex items-center gap-2">
50		− <code class="bg-base-300 px-3 py-2 rounded text-sm flex-grow">/apollos/keryx/<%= @keryx.name %>/control</code>
	50	+ <code class="bg-base-300 px-3 py-2 rounded text-sm flex-grow"><%= @mqtt_base %>/control</code>
51	51	<button
52	52	class="btn btn-square btn-sm"
53	53	phx-click={JS.dispatch("phx:copy", to: "#control-topic")}
54	54	title="Copy control topic">
55	55	<i class="fas fa-copy"></i>
56	56	</button>
57		− <input type="hidden" id="control-topic" value={"/apollos/keryx/#{@keryx.name}/control"} />
	57	+ <input type="hidden" id="control-topic" value={"#{@mqtt_base}/control"} />
58	58	</div>
59	59	<div class="mt-3 text-xs">
60	60	<p class="font-bold mb-1">Commands:</p>
61		− <code class="block bg-base-300 p-2 rounded mb-1">&left"action": "refresh"}</code>
62		− <code class="block bg-base-300 p-2 rounded">"action": "refresh", "query_id": 123}</code>
	61	+ <code class="block bg-base-300 p-2 rounded mb-1">{"action": "refresh"}</code>
	62	+ <code class="block bg-base-300 p-2 rounded">{"action": "refresh", "query_id": 123}</code>
63	63	</div>
64	64	</div>
65	65	</div>

@@ -78,14 +78,14 @@

78	78	<% end %>
79	79	</div>
80	80	<div class="flex items-center gap-2">
81		− <code class="bg-base-300 px-3 py-2 rounded text-sm flex-grow">/apollos/keryx/<%= @keryx.name %>/<%= q.source.type %>/<%= q.id %></code>
	81	+ <code class="bg-base-300 px-3 py-2 rounded text-sm flex-grow"><%= @mqtt_base %>/<%= q.source.type %>/<%= q.id %></code>
82	82	<button
83	83	class="btn btn-square btn-sm"
84	84	phx-click={JS.dispatch("phx:copy", to: "#topic-#{q.id}")}
85	85	title="Copy topic">
86	86	<i class="fas fa-copy"></i>
87	87	</button>
88		− <input type="hidden" id={"topic-#{q.id}"} value={["/apollos/keryx/", @keryx.name, q.source.type, q.id] \|> Enum.join("/")} />
	88	+ <input type="hidden" id={"topic-#{q.id}"} value={[@mqtt_base, q.source.type, q.id] \|> Enum.join("/")} />
89	89	</div>
90	90	</div>
91	91	<div class="flex justify-between items-center">

@@ -116,14 +116,14 @@

116	116	<div class="ml-4 mt-2">
117	117	<p class="mb-2">
118	118	<span class="font-semibold">Metadata:</span>
119		− <code class="bg-base-200 px-2 py-1 rounded ml-1">/apollos/keryx/<%= @keryx.name %>/<%= q.source.type %>/<%= q.id %>/meta</code>
	119	+ <code class="bg-base-200 px-2 py-1 rounded ml-1"><%= @mqtt_base %>/<%= q.source.type %>/<%= q.id %>/meta</code>
120	120	<span class="block ml-2 mt-1 text-[10px] opacity-60">
121	121	Contains condensed data summary and query information (name, metadata)
122	122	</span>
123	123	</p>
124	124	<p class="mb-1">
125	125	<span class="font-semibold">Individual items:</span>
126		− <code class="bg-base-200 px-2 py-1 rounded ml-1">/apollos/keryx/<%= @keryx.name %>/<%= q.source.type %>/<%= q.id %>/[item-id]</code>
	126	+ <code class="bg-base-200 px-2 py-1 rounded ml-1"><%= @mqtt_base %>/<%= q.source.type %>/<%= q.id %>/[item-id]</code>
127	127	</p>
128	128	<p class="text-xs opacity-60">
129	129	<%= case q.source.type do %>

@@ -164,8 +164,8 @@

164	164	<span class="text-[10px] font-semibold">All items for route</span>
165	165	</div>
166	166	<div class="flex items-center gap-2 opacity-60">
167		− <code class="bg-base-300 px-2 py-1 rounded text-[10px] flex-grow truncate">/apollos/keryx/<%= @keryx.name %>/<%= q.source.type %>/<%= q.id %>/<%= route %></code>
168		− <input type="hidden" id={"route-topic-#{q.id}-#{route}"} value={"/apollos/keryx/#{@keryx.name}/#{q.source.type}/#{q.id}/#{route}"} />
	167	+ <code class="bg-base-300 px-2 py-1 rounded text-[10px] flex-grow truncate"><%= @mqtt_base %>/<%= q.source.type %>/<%= q.id %>/<%= route %></code>
	168	+ <input type="hidden" id={"route-topic-#{q.id}-#{route}"} value={"#{@mqtt_base}/#{q.source.type}/#{q.id}/#{route}"} />
169	169	<div class="flex gap-0.5 flex-shrink-0">
170	170	<button
171	171	class="btn btn-ghost btn-xs p-0 min-h-0 h-4 w-4"

@@ -190,8 +190,8 @@

190	190	</div>
191	191	</div>
192	192	<div class="flex items-center gap-2 opacity-75">
193		− <code class="bg-base-300 px-2 py-1 rounded text-[10px] flex-grow truncate">/apollos/keryx/<%= @keryx.name %>/<%= q.source.type %>/<%= q.id %>/<%= group.route %>/<%= group.destination %></code>
194		− <input type="hidden" id={"mini-topic-#{q.id}-#{route}-#{idx}"} value={"/apollos/keryx/#{@keryx.name}/#{q.source.type}/#{q.id}/#{group.route}/#{group.destination}"} />
	193	+ <code class="bg-base-300 px-2 py-1 rounded text-[10px] flex-grow truncate"><%= @mqtt_base %>/<%= q.source.type %>/<%= q.id %>/<%= group.route %>/<%= group.destination %></code>
	194	+ <input type="hidden" id={"mini-topic-#{q.id}-#{route}-#{idx}"} value={"#{@mqtt_base}/#{q.source.type}/#{q.id}/#{group.route}/#{group.destination}"} />
195	195	<div class="flex gap-0.5 flex-shrink-0">
196	196	<button
197	197	class="btn btn-ghost btn-xs p-0 min-h-0 h-4 w-4"

@@ -232,8 +232,8 @@

232	232	</div>
233	233	</div>
234	234	<div class="flex items-center gap-2 opacity-75">
235		− <code class="bg-base-300 px-2 py-1 rounded text-[10px] flex-grow truncate">/apollos/keryx/<%= @keryx.name %>/<%= q.source.type %>/<%= q.id %>/<%= group.id %></code>
236		− <input type="hidden" id={"mini-topic-#{q.id}-#{idx}"} value={"/apollos/keryx/#{@keryx.name}/#{q.source.type}/#{q.id}/#{group.id}"} />
	235	+ <code class="bg-base-300 px-2 py-1 rounded text-[10px] flex-grow truncate"><%= @mqtt_base %>/<%= q.source.type %>/<%= q.id %>/<%= group.id %></code>
	236	+ <input type="hidden" id={"mini-topic-#{q.id}-#{idx}"} value={"#{@mqtt_base}/#{q.source.type}/#{q.id}/#{group.id}"} />
237	237	<div class="flex gap-0.5 flex-shrink-0">
238	238	<button
239	239	class="btn btn-ghost btn-xs p-0 min-h-0 h-4 w-4"

modified apps/room_sanctum/test/room_sanctum_web/live/keryx_show_test.exs

+9 −2

@@ -1,6 +1,6 @@

1	1	defmodule RoomSanctumWeb.KeryxShowTest do
2	2	@moduledoc """
3		− The Keryx page, and whatever its worker has seen so far.
	3	+ The Keryx page: its topics, and whatever its worker has seen so far.
4	4
5	5	Builds its own fixtures, as SourceLiveBulkPaintTest does, since
6	6	ConfigurationFixtures predates the required user_id.

@@ -45,7 +45,14 @@ defmodule RoomSanctumWeb.KeryxShowTest do

45	45	user_id: user.id
46	46	})
47	47
48		− %{conn: log_in_user(conn, user), keryx: keryx}
	48	+ %{conn: log_in_user(conn, user), keryx: keryx, query: query}
	49	+ end
	50	+
	51	+ test "shows the topics a client subscribes to", %{conn: conn, keryx: keryx, query: query} do
	52	+ {:ok, _live, html} = live(conn, "/cfg/keryxiae/#{keryx.id}")
	53	+
	54	+ assert html =~ "apollos/keryx/#{keryx.id}/control"
	55	+ assert html =~ "apollos/keryx/#{keryx.id}/bourse/#{query.id}"
49	56	end
50	57
51	58	# The page asks the Keryx's item cache what it has seen. That cache is a

Parents: b1b7c62