hugo/bDS2
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: domain event bus for multi-client sync (issue #26, phase 2)

Browse Source
This commit is contained in:
Chili Palmer
2026-07-14 19:33:07 +02:00
parent 700ed79c43
commit 16f45d77b2
14 changed files with 309 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
lib/bds/desktop/shell_data.ex
View File

@@ -91,14 +91,16 @@ defmodule BDS.Desktop.ShellData do
end
end
def editor_meta(task_status) do
# Accepts the caller's already-resolved locale so hot render paths never
# re-read the ui.language setting from the database.
def editor_meta(task_status, ui_language \\ nil) do
[
%{
label: dgettext("ui", "Status"),
value: task_status.running_task_message || dgettext("ui", "Idle")
},
%{label: dgettext("ui", "Mode"), value: dgettext("ui", "Offline")},
%{label: dgettext("ui", "Main Language"), value: ui_language()}
%{label: dgettext("ui", "Main Language"), value: ui_language || ui_language()}
]
end
@@ -107,7 +109,7 @@ defmodule BDS.Desktop.ShellData do
post_count: dashboard.post_stats.total_posts,
media_count: dashboard.media_stats.media_count,
theme_badge: "desktop-shell",
ui_language: Keyword.get(opts, :ui_language, ui_language()),
ui_language: Keyword.get_lazy(opts, :ui_language, &ui_language/0),
offline_mode: Keyword.get(opts, :offline_mode, true),
running_task_message: task_status.running_task_message,
running_task_overflow: task_status.running_task_overflow,

16
lib/bds/desktop/shell_live.ex
View File

@@ -6,7 +6,6 @@ defmodule BDS.Desktop.ShellLive do
import Phoenix.HTML
alias BDS.{AI, Blogmark, BoundedAtoms, Metadata}
alias BDS.CliSync.Watcher
alias BDS.Desktop.{FilePicker, FolderPicker, ShellData, UILocale}
alias BDS.Desktop.ShellLive.{
@@ -151,7 +150,9 @@ defmodule BDS.Desktop.ShellLive do
connected = connected?(socket)
if connected do
Phoenix.PubSub.subscribe(BDS.PubSub, Watcher.topic())
# Entity + settings events from every source: in-app contexts, other
# connected clients (GUI or TUI), and the CLI sync watcher.
:ok = BDS.Events.subscribe()
Process.send_after(self(), :refresh_task_status, @refresh_interval)
end
@@ -736,6 +737,14 @@ defmodule BDS.Desktop.ShellLive do
{:noreply, assign(socket, :shell_overlay, overlay)}
end
# The UI language is a server-side setting shared by all clients; when any
# client (or script) changes it, every shell re-renders in the new locale.
def handle_info({:settings_changed, "ui.language"}, socket) do
{:noreply, set_page_language(socket, ShellData.ui_language())}
end
def handle_info({:settings_changed, _key}, socket), do: {:noreply, socket}
def handle_info(message, socket) do
Bridges.handle_info(message, socket, bridges_callbacks())
end
@@ -981,6 +990,9 @@ defmodule BDS.Desktop.ShellLive do
socket
else
UILocale.put(normalized)
# Persist server-side so every client renders the same language; the
# resulting settings_changed broadcast is a no-op for this socket.
_ = BDS.Settings.put_global_setting("ui.language", normalized)
socket
|> assign(:page_language, normalized)

2
lib/bds/desktop/shell_live/bridges.ex
View File

@@ -266,7 +266,7 @@ defmodule BDS.Desktop.ShellLive.Bridges do
socket
|> assign(:task_status, task_status)
|> assign(:editor_meta, ShellData.editor_meta(task_status))
|> assign(:editor_meta, ShellData.editor_meta(task_status, socket.assigns.page_language))
|> assign(
:status,
ShellData.status_bar(

2
lib/bds/desktop/shell_live/socket_state.ex
View File

@@ -38,7 +38,7 @@ defmodule BDS.Desktop.ShellLive.SocketState do
|> assign(:sidebar_header, active_sidebar_label(activity_buttons, workbench.active_view, sidebar_data))
|> assign(:panel_tabs, ShellData.panel_tabs(workbench))
|> assign(:current_tab, current_tab)
|> assign(:editor_meta, ShellData.editor_meta(task_status))
|> assign(:editor_meta, ShellData.editor_meta(task_status, page_language))
|> assign(
:status,
ShellData.status_bar(workbench, task_status, dashboard,

58
lib/bds/events.ex Normal file
View File

@@ -0,0 +1,58 @@
defmodule BDS.Events do
@moduledoc """
Domain event bus (issue #26, phase 2).
Contexts broadcast entity mutations on the same `"entity:changed"` topic
(and payload shape) that `BDS.CliSync.Watcher` already uses for external
CLI writes, so every connected client — LiveView shells and TUI sessions
alike — stays synchronized through one subscription regardless of where
a change originated.
"""
@entity_topic "entity:changed"
@settings_topic "settings:changed"
@type action :: :created | :updated | :deleted
@doc "Subscribe the calling process to entity and settings changes."
@spec subscribe() :: :ok
def subscribe do
:ok = Phoenix.PubSub.subscribe(BDS.PubSub, @entity_topic)
:ok = Phoenix.PubSub.subscribe(BDS.PubSub, @settings_topic)
end
@doc """
Broadcast an entity mutation as `{:entity_changed, %{entity: entity,
entity_id: id, action: action}}` — the same message the CLI sync watcher
emits, so subscribers handle both through one code path.
"""
@spec entity_changed(String.t(), String.t(), action()) :: :ok
def entity_changed(entity, entity_id, action)
when is_binary(entity) and is_binary(entity_id) and
action in [:created, :updated, :deleted] do
Phoenix.PubSub.broadcast(
BDS.PubSub,
@entity_topic,
{:entity_changed, %{entity: entity, entity_id: entity_id, action: action}}
)
end
@doc """
Pipe-friendly broadcast: on `{:ok, %{id: id}}` broadcasts `action` for
`entity` and returns the result unchanged; any other result passes
through untouched.
"""
@spec broadcast_result(term(), String.t(), action()) :: term()
def broadcast_result({:ok, %{id: id}} = result, entity, action) when is_binary(id) do
:ok = entity_changed(entity, id, action)
result
end
def broadcast_result(result, _entity, _action), do: result
@doc "Broadcast a global setting change as `{:settings_changed, key}`."
@spec settings_changed(String.t()) :: :ok
def settings_changed(key) when is_binary(key) do
Phoenix.PubSub.broadcast(BDS.PubSub, @settings_topic, {:settings_changed, key})
end
end

15
lib/bds/i18n.ex
View File

@@ -62,10 +62,23 @@ defmodule BDS.I18n do
def resolve_ui_locale(locale), do: resolve_supported_locale(locale) || @default_language
def current_ui_locale do
(System.get_env("LC_ALL") || System.get_env("LC_MESSAGES") || System.get_env("LANG"))
(stored_ui_locale() || System.get_env("LC_ALL") || System.get_env("LC_MESSAGES") ||
System.get_env("LANG"))
|> resolve_ui_locale()
end
# The UI language is a server-side setting shared by all connected clients
# (issue #26); the OS locale is only the fallback for a fresh install.
defp stored_ui_locale do
with true <- BDS.Repo.ready?(),
value when is_binary(value) and value != "" <-
BDS.Settings.get_global_setting("ui.language") do
value
else
_other -> nil
end
end
def resolve_render_locale(language), do: resolve_supported_locale(language) || @default_language
def format_locale(language) do

4
lib/bds/media.ex
View File

@@ -34,6 +34,7 @@ defmodule BDS.Media do
import Ecto.Query
alias BDS.Events
alias BDS.Media.Media
alias BDS.Media.Translation
alias BDS.Persistence
@@ -121,6 +122,7 @@ defmodule BDS.Media do
log_sidecar_error(write_sidecar(project, media), media.id)
log_thumbnail_error(ensure_thumbnails(project, media), media.id)
:ok = Search.sync_media(media)
:ok = Events.entity_changed("media", media.id, :created)
{:ok, media}
{:error, reason} ->
@@ -162,6 +164,7 @@ defmodule BDS.Media do
{:ok, updated_media} ->
log_sidecar_error(write_sidecar(project, updated_media), updated_media.id)
:ok = Search.sync_media(updated_media)
:ok = Events.entity_changed("media", updated_media.id, :updated)
{:ok, updated_media}
{:error, reason} ->
@@ -211,6 +214,7 @@ defmodule BDS.Media do
end)
Search.delete_media(media.id)
:ok = Events.entity_changed("media", media.id, :deleted)
{:ok, :deleted}
{:error, reason} ->

5
lib/bds/posts.ex
View File

@@ -12,6 +12,7 @@ defmodule BDS.Posts do
import BDS.MapUtils, only: [attr: 2, maybe_put: 3]
alias BDS.Embeddings
alias BDS.Events
alias BDS.Media
alias BDS.Persistence
alias BDS.PostLinks
@@ -106,6 +107,7 @@ defmodule BDS.Posts do
{:ok, post} ->
:ok = Embeddings.sync_post(post)
:ok = Search.sync_post(post)
:ok = Events.entity_changed("post", post.id, :created)
{:ok, post}
error ->
@@ -151,6 +153,7 @@ defmodule BDS.Posts do
:ok = AutoTranslation.maybe_schedule(updated_post)
end
:ok = Events.entity_changed("post", updated_post.id, :updated)
{:ok, updated_post}
error ->
@@ -202,6 +205,7 @@ defmodule BDS.Posts do
:ok = Translations.publish_post_translations(updated_post)
:ok = PostLinks.sync_post_links(updated_post)
:ok = Search.sync_post(updated_post)
:ok = Events.entity_changed("post", updated_post.id, :updated)
{:ok, updated_post}
error ->
@@ -335,6 +339,7 @@ defmodule BDS.Posts do
PostLinks.delete_post_links(deleted_post.id)
Enum.each(linked_media_ids, &sync_deleted_post_media_sidecar/1)
Search.delete_post(deleted_post.id)
:ok = Events.entity_changed("post", deleted_post.id, :deleted)
{:ok, :deleted}
{:error, changeset} ->

6
lib/bds/scripts.ex
View File

@@ -5,6 +5,7 @@ defmodule BDS.Scripts do
import BDS.MapUtils, only: [attr: 2, maybe_put: 3]
alias BDS.DocumentFields
alias BDS.Events
alias BDS.Frontmatter
alias BDS.Persistence
alias BDS.ProgressReporter
@@ -40,6 +41,7 @@ defmodule BDS.Scripts do
updated_at: now
})
|> Repo.insert()
|> Events.broadcast_result("script", :created)
end
@spec create_and_publish_script(attrs()) :: script_result()
@@ -84,6 +86,7 @@ defmodule BDS.Scripts do
serialize_script_file(script, content)
)
:ok = Events.entity_changed("script", script.id, :created)
{:ok, script}
end
@@ -132,6 +135,7 @@ defmodule BDS.Scripts do
updated_at: updated_at
})
|> Repo.update()
|> Events.broadcast_result("script", :updated)
{:error, reason} ->
{:error, {:invalid_script, reason}}
@@ -176,6 +180,7 @@ defmodule BDS.Scripts do
script
|> Script.changeset(updates)
|> Repo.update()
|> Events.broadcast_result("script", :updated)
end
end
@@ -189,6 +194,7 @@ defmodule BDS.Scripts do
case Repo.delete(script) do
{:ok, deleted_script} ->
delete_file_if_present(deleted_script.project_id, deleted_script.file_path)
:ok = Events.entity_changed("script", deleted_script.id, :deleted)
{:ok, :deleted}
{:error, changeset} ->

8
lib/bds/settings.ex
View File

@@ -27,8 +27,12 @@ defmodule BDS.Settings do
})
|> Repo.insert_or_update()
|> case do
{:ok, _setting} -> :ok
{:error, reason} -> {:error, reason}
{:ok, _setting} ->
:ok = BDS.Events.settings_changed(key)
:ok
{:error, reason} ->
{:error, reason}
end
end
end

7
lib/bds/tags.ex
View File

@@ -6,6 +6,7 @@ defmodule BDS.Tags do
import Ecto.Query
alias BDS.Events
alias BDS.MapUtils
alias BDS.Persistence
alias BDS.Posts
@@ -40,6 +41,7 @@ defmodule BDS.Tags do
|> case do
{:ok, tag} ->
with :ok <- write_tags_json(project_id) do
:ok = Events.entity_changed("tag", tag.id, :created)
{:ok, tag}
end
@@ -130,6 +132,7 @@ defmodule BDS.Tags do
|> case do
{:ok, updated_tag} ->
with :ok <- write_tags_json(updated_tag.project_id) do
:ok = Events.entity_changed("tag", updated_tag.id, :updated)
{:ok, updated_tag}
end
@@ -165,6 +168,7 @@ defmodule BDS.Tags do
{:ok, post_ids} ->
with :ok <- rewrite_published_posts(post_ids),
:ok <- write_tags_json(tag.project_id) do
:ok = Events.entity_changed("tag", tag.id, :deleted)
{:ok, :deleted}
end
@@ -205,6 +209,7 @@ defmodule BDS.Tags do
{:ok, {updated_tag, post_ids}} ->
with :ok <- rewrite_published_posts(post_ids),
:ok <- write_tags_json(tag.project_id) do
:ok = Events.entity_changed("tag", updated_tag.id, :updated)
{:ok, updated_tag}
end
@@ -250,6 +255,8 @@ defmodule BDS.Tags do
{:ok, post_ids} ->
with :ok <- rewrite_published_posts(post_ids),
:ok <- write_tags_json(target_tag.project_id) do
Enum.each(source_tags, &Events.entity_changed("tag", &1.id, :deleted))
:ok = Events.entity_changed("tag", target_tag.id, :updated)
{:ok, :merged}
end

6
lib/bds/templates.ex
View File

@@ -17,6 +17,7 @@ defmodule BDS.Templates do
alias BDS.Projects
alias BDS.Repo
alias BDS.Slug
alias BDS.Events
alias BDS.Tags
alias BDS.Templates.Template
@@ -58,6 +59,7 @@ defmodule BDS.Templates do
serialize_template_file(template, template.content || "")
)
:ok = Events.entity_changed("template", template.id, :created)
{:ok, template}
end
end
@@ -102,6 +104,7 @@ defmodule BDS.Templates do
updated_at: updated_at
})
|> Repo.update()
|> Events.broadcast_result("template", :updated)
{:error, reason} ->
{:error, {:invalid_liquid, reason}}
@@ -149,6 +152,7 @@ defmodule BDS.Templates do
serialize_template_file(template, content)
)
:ok = Events.entity_changed("template", template.id, :created)
{:ok, template}
end
@@ -183,6 +187,7 @@ defmodule BDS.Templates do
affected_posts,
slug_changed?
) do
:ok = Events.entity_changed("template", updated_template.id, :updated)
{:ok, updated_template}
end
end
@@ -299,6 +304,7 @@ defmodule BDS.Templates do
case Repo.delete(template) do
{:ok, deleted_template} ->
delete_file_if_present(deleted_template.project_id, deleted_template.file_path)
:ok = Events.entity_changed("template", deleted_template.id, :deleted)
{:ok, :deleted}
{:error, changeset} ->

59
specs/events.allium Normal file
View File

@@ -0,0 +1,59 @@
-- allium: 1
-- Domain Event Bus (issue #26, phase 2)
-- Scope: extension (multi-client synchronization over Phoenix.PubSub)
-- Distilled from: lib/bds/events.ex, context broadcast call sites,
-- lib/bds/desktop/shell_live.ex, lib/bds/cli_sync/watcher.ex
entity EntityChangedEvent {
entity: String -- post, media, tag, template, script
entity_id: String
action: created | updated | deleted
-- Same topic and payload shape as the CLI sync watcher, so one
-- subscription covers in-app mutations and external CLI writes.
}
entity SettingsChangedEvent {
key: String -- e.g. ui.language
}
surface EventBusSurface {
facing _: EventBus
provides:
ContextMutationSucceeded(entity, entity_id, action)
GlobalSettingWritten(key)
ClientSubscribed()
}
rule BroadcastEntityMutation {
when: ContextMutationSucceeded(entity, entity_id, action)
-- Posts, Media, Tags, Templates, Scripts broadcast after every
-- successful create/update/publish/delete (publish maps to updated).
ensures: EntityChangedEvent.created(
entity: entity,
entity_id: entity_id,
action: action
)
}
rule BroadcastSettingsChange {
when: GlobalSettingWritten(key)
ensures: SettingsChangedEvent.created(key: key)
}
rule ClientSynchronization {
when: ClientSubscribed()
-- Every shell (LiveView GUI or TUI session) subscribes on mount and
-- refreshes affected views on each event; deleted entities close
-- their open tabs. This keeps all connected clients synchronized
-- regardless of which client or pipeline originated the change.
ensures: ClientRefreshOnEvent()
}
rule ServerSideUiLanguage {
when: GlobalSettingWritten(key: "ui.language")
-- The UI language is a single server-side setting: changing it in any
-- client persists it and re-renders every connected client. The OS
-- locale is only the fallback when the setting is unset.
ensures: AllClientsRelocalized()
}

123
test/bds/events_test.exs Normal file
View File

@@ -0,0 +1,123 @@
defmodule BDS.EventsTest do
use ExUnit.Case, async: false
alias BDS.Events
setup do
:ok = Ecto.Adapters.SQL.Sandbox.checkout(BDS.Repo)
:ok = Events.subscribe()
temp_dir = Path.join(System.tmp_dir!(), "bds-events-#{System.unique_integer([:positive])}")
File.mkdir_p!(temp_dir)
on_exit(fn -> File.rm_rf(temp_dir) end)
{:ok, project} = BDS.Projects.create_project(%{name: "Events", data_path: temp_dir})
%{project: project}
end
describe "entity_changed/3" do
test "broadcasts the watcher-compatible payload" do
:ok = Events.entity_changed("post", "abc", :updated)
assert_receive {:entity_changed, %{entity: "post", entity_id: "abc", action: :updated}}
end
end
describe "posts broadcast" do
test "create, update, publish, delete", %{project: project} do
{:ok, post} = BDS.Posts.create_post(%{project_id: project.id, title: "Hello"})
post_id = post.id
assert_receive {:entity_changed, %{entity: "post", entity_id: ^post_id, action: :created}}
{:ok, _} = BDS.Posts.update_post(post_id, %{title: "Hello 2"})
assert_receive {:entity_changed, %{entity: "post", entity_id: ^post_id, action: :updated}}
{:ok, _} = BDS.Posts.publish_post(post_id)
assert_receive {:entity_changed, %{entity: "post", entity_id: ^post_id, action: :updated}}
{:ok, _} = BDS.Posts.delete_post(post_id)
assert_receive {:entity_changed, %{entity: "post", entity_id: ^post_id, action: :deleted}}
end
end
describe "tags broadcast" do
test "create, update, delete", %{project: project} do
{:ok, tag} = BDS.Tags.create_tag(%{project_id: project.id, name: "elixir"})
tag_id = tag.id
assert_receive {:entity_changed, %{entity: "tag", entity_id: ^tag_id, action: :created}}
{:ok, _} = BDS.Tags.update_tag(tag_id, %{name: "erlang"})
assert_receive {:entity_changed, %{entity: "tag", entity_id: ^tag_id, action: :updated}}
{:ok, _} = BDS.Tags.delete_tag(tag_id)
assert_receive {:entity_changed, %{entity: "tag", entity_id: ^tag_id, action: :deleted}}
end
end
describe "templates broadcast" do
test "create, update, delete", %{project: project} do
{:ok, template} =
BDS.Templates.create_template(%{
project_id: project.id,
title: "Article",
kind: :post,
content: "<article>{{ content }}</article>"
})
template_id = template.id
assert_receive {:entity_changed,
%{entity: "template", entity_id: ^template_id, action: :created}}
{:ok, _} = BDS.Templates.update_template(template_id, %{title: "Article 2"})
assert_receive {:entity_changed,
%{entity: "template", entity_id: ^template_id, action: :updated}}
{:ok, _} = BDS.Templates.delete_template(template_id)
assert_receive {:entity_changed,
%{entity: "template", entity_id: ^template_id, action: :deleted}}
end
end
describe "scripts broadcast" do
test "create, update, delete", %{project: project} do
{:ok, script} =
BDS.Scripts.create_script(%{
project_id: project.id,
title: "hello",
kind: :utility,
content: "function main() end"
})
script_id = script.id
assert_receive {:entity_changed, %{entity: "script", entity_id: ^script_id, action: :created}}
{:ok, _} = BDS.Scripts.update_script(script_id, %{title: "hello2"})
assert_receive {:entity_changed, %{entity: "script", entity_id: ^script_id, action: :updated}}
{:ok, _} = BDS.Scripts.delete_script(script_id)
assert_receive {:entity_changed, %{entity: "script", entity_id: ^script_id, action: :deleted}}
end
end
describe "settings broadcast" do
test "put_global_setting broadcasts settings_changed" do
:ok = BDS.Settings.put_global_setting("ui.git_diff_view_style", "split")
assert_receive {:settings_changed, "ui.git_diff_view_style"}
end
test "ui.language setting overrides the OS locale server-side" do
:ok = BDS.Settings.put_global_setting("ui.language", "de")
assert_receive {:settings_changed, "ui.language"}
assert BDS.I18n.current_ui_locale() == "de"
# Blank value falls back to the OS locale (en in the test env).
:ok = BDS.Settings.put_global_setting("ui.language", "")
assert BDS.I18n.current_ui_locale() == "en"
end
end
end