feat: scaffold Phase B + C (events, telemetry, plugins, Solid.js)

Phase B Step 6 — Solid.js incremental migration
  - vite-plugin-solid + solid-js + @solidjs/testing-library in package.json
  - vite.config.ts compiles src/components/solid/** as Solid TSX
  - tsconfig.json gains jsx: preserve / jsxImportSource: solid-js
  - lib/solidAdapter.ts wraps existing custom Stores as Solid signals
  - lib/solidMount.ts adapts Solid render to {mount,destroy} contract
  - components/solid/Badge.tsx (proof-of-concept leaf)
  - components/solid/ChannelListItem.tsx (store-subscribed leaf)
  - components/solid/Badge.test.tsx pipeline smoke test
  - components/solid/README.md documents the migration recipe

Phase B Step 7 — Event persistence layer
  - SQLite + Postgres migrations for the events table
  - sqlc query files for both engines
  - EventStore interface + SQLite raw-SQL impl + MemStore impl + pg stubs
  - ws.EventPersister: async batched writer (queue / flush / drain / drop)
  - ws.StartEventPruner: background retention pruner
  - hub persists every replay-buffer push and exposes reconnect-tier counters
  - serve.handleReconnect: tiered replay (buffer -> DB -> full re-sync)
  - EventPersistenceConfig + main.go wiring
  - event_persister_test.go covers batching / drops / drain

Phase B Step 8 — OpenTelemetry skeleton
  - Server/telemetry package with public Provider/Tracer/Meter/Counter API
  - telemetry_default.go (no-op build) + telemetry_otel.go (build tag otel)
  - telemetry/metrics.go declares the AppMetrics bundle
  - HTTPMiddleware mounted in Chi router (pass-through in default build)
  - PrometheusHandler optionally mounted at /metrics
  - Spans on MessageService.SendMessage, PermissionService.HasChannelPerm,
    ChannelService.ListVisibleChannels
  - Reconnect-tier counter wired into the global meter
  - TelemetryConfig defaults

Phase C Step 9 — Wazero plugin runtime skeleton
  - Server/plugin package: manifest parser, loader, registry, host APIs
    (commands, storage, events, http, ui), errors
  - sandbox_default.go (no-op) + sandbox_wazero.go (build tag wazero)
  - SQLite + Postgres migrations for plugins + plugin_kv tables
  - PluginStore interface + impls + pg stubs
  - plugin/examples/hello manifest + README
  - plugin_test.go covers manifest, loader, capability gating
  - api/plugins_handler.go admin REST surface, mounted under admin group
  - PluginsConfig + main.go wiring (disabled by default)
  - Client: lib/pluginBridge.ts iframe + postMessage host
  - Client: components/solid/PluginContainer.tsx Solid host component

Verification
  - Default build (no -tags) is intended to compile cleanly with no new
    third-party dependencies. The sandbox lacked Go 1.25.0 so go build
    could not run; PHASE_BC_LOCAL_TODO.md enumerates the local follow-up
    work (npm install, go mod tidy, sqlc-generate, real otel/wazero
    wiring, remaining service spans, full Solid migration).
This commit is contained in:
Claude
2026-04-06 09:00:47 +00:00
parent 8b49e64581
commit 34b6bf3217
57 changed files with 3798 additions and 9 deletions
+3
View File
@@ -32,6 +32,7 @@
"devDependencies": {
"@eslint/js": "^9.39.4",
"@playwright/test": "^1",
"@solidjs/testing-library": "^0.8.10",
"@stryker-mutator/core": "^9.6.0",
"@stryker-mutator/typescript-checker": "^9.6.0",
"@stryker-mutator/vitest-runner": "^9.6.0",
@@ -46,6 +47,7 @@
"typescript": "^5.7",
"typescript-eslint": "^8.58.0",
"vite": "^6",
"vite-plugin-solid": "^2.11.0",
"vitest": "^3"
},
"prettier": {
@@ -70,6 +72,7 @@
"@tauri-apps/plugin-store": "^2",
"@tauri-apps/plugin-updater": "^2.10.0",
"livekit-client": "^2.18.0",
"solid-js": "^1.9.3",
"zod": "^4.3.6"
}
}
@@ -0,0 +1,27 @@
/**
* Phase B Step 6 — Solid pipeline smoke test.
*
* Verifies that the Vite + Solid + Vitest configuration actually compiles
* and renders a component end-to-end. This test only exists to prove the
* pipeline; the test scope expands as more components migrate.
*/
import { describe, it, expect } from "vitest";
import { render } from "@solidjs/testing-library";
import { Badge } from "./Badge";
describe("Badge (solid)", () => {
it("renders the label", () => {
const { getByText } = render(() => <Badge label="online" variant="online" />);
expect(getByText("online")).toBeTruthy();
});
it("invokes onClick when activated", () => {
let clicked = 0;
const { getByText } = render(() => (
<Badge label="press me" onClick={() => clicked++} />
));
getByText("press me").click();
expect(clicked).toBe(1);
});
});
@@ -0,0 +1,46 @@
/**
* Phase B Step 6 — first Solid.js leaf component.
*
* Trivial proof-of-concept badge used by other Solid components and the
* mount helper. Self-contained: no store subscriptions, no async work, just
* a presentational element. Use it as the canonical example when migrating
* vanilla badge/pill components in the rest of the tree.
*/
import type { JSX } from "solid-js";
export interface BadgeProps {
label: string;
/** Visual variant; defaults to "neutral". */
variant?: "neutral" | "online" | "idle" | "dnd" | "offline";
/** Optional click handler. When set the badge gains role="button". */
onClick?: () => void;
}
const variantClass: Record<NonNullable<BadgeProps["variant"]>, string> = {
neutral: "badge",
online: "badge badge--online",
idle: "badge badge--idle",
dnd: "badge badge--dnd",
offline: "badge badge--offline",
};
export function Badge(props: BadgeProps): JSX.Element {
const cls = () => variantClass[props.variant ?? "neutral"];
return (
<span
class={cls()}
role={props.onClick ? "button" : undefined}
tabIndex={props.onClick ? 0 : undefined}
onClick={props.onClick}
onKeyDown={(e) => {
if (props.onClick && (e.key === "Enter" || e.key === " ")) {
e.preventDefault();
props.onClick();
}
}}
>
{props.label}
</span>
);
}
@@ -0,0 +1,52 @@
/**
* Phase B Step 6 — second Solid.js leaf component.
*
* Renders a single row in the channel list. Reads from the existing
* `channelsStore` via the Solid adapter so it stays in sync with whatever
* the vanilla dispatcher writes into the store.
*
* This is the canonical example for migrating list-item components: pure
* presentation, fed by an accessor, with a click handler delegated up to the
* parent so the component knows nothing about the dispatcher.
*/
import type { JSX } from "solid-js";
import { Show } from "solid-js";
import { fromStoreSlice } from "@lib/solidAdapter";
import { channelsStore, type Channel, type ChannelsState } from "@stores/channels.store";
import { Badge } from "./Badge";
export interface ChannelListItemProps {
channelId: number;
onSelect: (id: number) => void;
}
export function ChannelListItem(props: ChannelListItemProps): JSX.Element {
// Subscribe to just this row's channel object so unrelated changes don't
// re-render. Falsy → row is hidden until the channel arrives.
const channel = fromStoreSlice<ChannelsState, Channel | undefined>(
channelsStore,
(s) => s.channels.get(props.channelId),
);
const isActive = fromStoreSlice<ChannelsState, boolean>(
channelsStore,
(s) => s.activeChannelId === props.channelId,
);
return (
<Show when={channel()}>
{(ch) => (
<li
class={"channel-list-item" + (isActive() ? " channel-list-item--active" : "")}
onClick={() => props.onSelect(props.channelId)}
>
<span class="channel-list-item__hash">#</span>
<span class="channel-list-item__name">{ch().name}</span>
<Show when={ch().unreadCount > 0}>
<Badge label={String(ch().unreadCount)} variant="dnd" />
</Show>
</li>
)}
</Show>
);
}
@@ -0,0 +1,30 @@
/**
* Phase C Step 9 — Solid component that hosts a plugin tab.
*
* Mounts the plugin's iframe via `pluginBridge.mount(...)` and tears it down
* when the component is disposed. The container itself is intentionally tiny:
* the bridge owns the iframe lifecycle and the postMessage protocol.
*/
import { onMount, onCleanup, type JSX } from "solid-js";
import { pluginBridge, type PluginTabBinding } from "@lib/pluginBridge";
export interface PluginContainerProps {
binding: PluginTabBinding;
}
export function PluginContainer(props: PluginContainerProps): JSX.Element {
let host!: HTMLDivElement;
let dispose: (() => void) | undefined;
onMount(() => {
dispose = pluginBridge.mount(props.binding, host);
});
onCleanup(() => {
dispose?.();
});
return (
<div class="plugin-container" data-plugin-id={props.binding.pluginId} ref={host} />
);
}
@@ -0,0 +1,49 @@
# Solid.js components
Phase B Step 6 lives here. This directory holds the incremental Solid.js
migration of the OwnCord client. Vanilla TypeScript components and Solid
components coexist throughout the migration; new UI work goes in this
directory, and existing leaf components are ported one PR at a time.
## Migration recipe
1. **Pick a leaf component.** Start with components that have no children of
their own and read at most one or two stores. Avoid container components
until every leaf inside them is Solid-native.
2. **Read state via the adapter.** Import `fromStore` or `fromStoreSlice` from
`@lib/solidAdapter` and pass the existing custom store. Do **not** rewrite
the store — Solid components and vanilla components share the same source
of truth.
3. **Mount via `solidMount`.** Containers that aren't yet Solid-native should
call `mountSolid(component, parentEl)` from `@lib/solidMount`. The returned
handle has the same `{ destroy }` shape that the rest of the codebase uses.
4. **Test the pipeline.** New components get a `*.test.tsx` next to them
using `@solidjs/testing-library`. The tests run under the existing Vitest
configuration without any extra setup.
5. **Delete vanilla DOM code.** Once a component is fully migrated, remove
the old factory function and update its callers to import from
`@components/solid/...`.
## Allowed reactivity
- `createSignal`, `createMemo`, `createEffect`, `createResource`
- `Show`, `For`, `Switch`/`Match`, `Index`
- `onMount`, `onCleanup`
## Forbidden patterns
- Direct DOM manipulation inside Solid components — use Solid's bindings or a
ref. The point of the migration is to delete manual DOM lifecycle code.
- Re-implementing existing stores in Solid's `createStore`. Wrap the existing
custom store via `fromStore` instead.
- Touching framework-agnostic code (`lib/ws.ts`, `lib/dispatcher.ts`,
`lib/livekitSession.ts`, `lib/api.ts`). These never need to know about Solid.
## Existing components
- `Badge.tsx` — presentational badge / pill (no store dependency)
- `ChannelListItem.tsx` — single channel row (subscribes to channels.store)
The plugin client bridge introduced by Phase C also adds a `PluginContainer`
component to this directory; see `Server/plugin/host_ui.go` for the host
side of that contract.
+110
View File
@@ -0,0 +1,110 @@
/**
* Phase C Step 9 — client-side plugin bridge.
*
* Mounts plugin UI tabs in sandboxed iframes and forwards postMessage traffic
* between the host client and each plugin. The host injects theme CSS
* variables on every load so plugin UIs match OwnCord's look and feel
* without each plugin re-implementing them.
*
* The bridge is intentionally tiny: it owns iframe lifecycles and message
* routing; everything else (rendering tabs, fetching the plugin list) lives
* in PluginContainer.tsx.
*/
export interface PluginTabBinding {
pluginId: number;
pluginName: string;
tabId: string;
label: string;
asset: string;
}
export interface PluginMessageEnvelope {
pluginId: number;
type: string;
payload?: unknown;
}
type Listener = (env: PluginMessageEnvelope) => void;
const HOST_ORIGIN_PREFIX = "owncord-plugin-host";
class PluginBridge {
private frames = new Map<number, HTMLIFrameElement>();
private listeners = new Set<Listener>();
private themeVars: Record<string, string> = {};
constructor() {
window.addEventListener("message", this.onMessage);
}
/** Replace the theme variables broadcast to plugin iframes. */
setTheme(vars: Record<string, string>): void {
this.themeVars = { ...vars };
for (const [pid, frame] of this.frames) {
this.postToFrame(pid, frame, { type: "theme", payload: this.themeVars });
}
}
/** Mount an iframe for binding into parent. Returns a destroy function. */
mount(binding: PluginTabBinding, parent: HTMLElement): () => void {
const iframe = document.createElement("iframe");
iframe.className = "plugin-iframe";
iframe.sandbox.add("allow-scripts");
iframe.title = `${binding.pluginName}: ${binding.label}`;
iframe.src = `/api/v1/plugins/${encodeURIComponent(binding.pluginName)}/ui/${binding.asset}`;
iframe.dataset.pluginId = String(binding.pluginId);
iframe.addEventListener("load", () => {
this.postToFrame(binding.pluginId, iframe, { type: "theme", payload: this.themeVars });
this.postToFrame(binding.pluginId, iframe, { type: "ready", payload: null });
});
parent.appendChild(iframe);
this.frames.set(binding.pluginId, iframe);
return () => {
iframe.remove();
this.frames.delete(binding.pluginId);
};
}
/** Listen for messages emitted by any mounted plugin iframe. */
onMessageEnvelope(listener: Listener): () => void {
this.listeners.add(listener);
return () => this.listeners.delete(listener);
}
/** Send a host → plugin message. */
send(pluginId: number, type: string, payload?: unknown): void {
const frame = this.frames.get(pluginId);
if (!frame) return;
this.postToFrame(pluginId, frame, { type, payload });
}
private postToFrame(pluginId: number, frame: HTMLIFrameElement, msg: { type: string; payload: unknown }): void {
frame.contentWindow?.postMessage(
{ source: HOST_ORIGIN_PREFIX, pluginId, ...msg },
"*",
);
}
private onMessage = (e: MessageEvent): void => {
const data = e.data;
if (!data || typeof data !== "object") return;
if ((data as { source?: unknown }).source === HOST_ORIGIN_PREFIX) return; // own echo
const env = data as { pluginId?: unknown; type?: unknown; payload?: unknown };
if (typeof env.pluginId !== "number" || typeof env.type !== "string") return;
const envelope: PluginMessageEnvelope = {
pluginId: env.pluginId,
type: env.type,
payload: env.payload,
};
for (const l of this.listeners) {
try {
l(envelope);
} catch (err) {
console.error("plugin bridge listener threw", err);
}
}
};
}
export const pluginBridge = new PluginBridge();
@@ -0,0 +1,56 @@
/**
* Phase B Step 6 — Solid.js adapter.
*
* Bridges the existing custom reactive `Store<T>` (lib/store.ts) into Solid's
* signal model so migrated components can read from existing stores without
* touching them. Vanilla and Solid components coexist throughout the
* migration: a vanilla component can update a store, and any Solid component
* subscribed via this adapter sees the new value through its accessor.
*
* Usage from a Solid component:
*
* import { fromStore } from "@lib/solidAdapter";
* import { authStore } from "@stores/auth";
*
* const auth = fromStore(authStore);
* return <div>{auth().username}</div>;
*
* The accessor returned by fromStore is a Solid signal getter, so it triggers
* fine-grained reactivity in any computation, JSX expression, or `createMemo`.
*/
import { createSignal, onCleanup, type Accessor } from "solid-js";
import type { Store } from "./store";
/**
* Wrap a custom Store as a Solid signal accessor. The signal updates whenever
* the underlying store fires, and the subscription is torn down when the
* Solid owner is disposed (so leaf components don't leak listeners).
*/
export function fromStore<T>(store: Store<T>): Accessor<T> {
const [value, setValue] = createSignal<T>(store.getState(), { equals: false });
const unsub = store.subscribe((next) => setValue(() => next));
onCleanup(unsub);
return value;
}
/**
* Wrap a derived slice of a store. Equivalent to fromStore(store).map(selector)
* but uses the store's native subscribeSelector so changes are gated by the
* existing equality comparator.
*/
export function fromStoreSlice<T, S>(
store: Store<T>,
selector: (state: T) => S,
isEqual?: (a: S, b: S) => boolean,
): Accessor<S> {
const initial = selector(store.getState());
const [value, setValue] = createSignal<S>(initial, { equals: false });
const unsub = store.subscribeSelector(
selector,
(next) => setValue(() => next),
isEqual,
);
onCleanup(unsub);
return value;
}
+38
View File
@@ -0,0 +1,38 @@
/**
* Phase B Step 6 — Solid.js mount helper.
*
* Wraps Solid's `render(...)` so a Solid component conforms to the
* `{ mount, destroy }` factory contract used everywhere else in the vanilla
* codebase. Existing container components can host a Solid leaf without
* being aware of Solid at all:
*
* import { mountSolid } from "@lib/solidMount";
* import { Badge } from "@components/solid/Badge";
*
* const handle = mountSolid(() => Badge({ label: "online" }), parentEl);
* // …later
* handle.destroy();
*/
import { render, type JSX } from "solid-js/web";
export interface SolidMount {
/** The DOM element the Solid root is rendered into. */
el: HTMLElement;
/** Tear the Solid root down and remove it from the DOM. */
destroy(): void;
}
export function mountSolid(component: () => JSX.Element, parent: HTMLElement): SolidMount {
const host = document.createElement("div");
host.dataset.solidRoot = "true";
parent.appendChild(host);
const dispose = render(component, host);
return {
el: host,
destroy() {
dispose();
host.remove();
},
};
}
+2
View File
@@ -11,6 +11,8 @@
"resolveJsonModule": true,
"isolatedModules": true,
"noEmit": true,
"jsx": "preserve",
"jsxImportSource": "solid-js",
"lib": [
"ES2023",
"DOM",
+12 -1
View File
@@ -1,5 +1,10 @@
import { defineConfig, type Plugin } from "vite";
import { resolve } from "path";
// Phase B Step 6 — Solid.js incremental migration. The plugin compiles
// JSX/TSX files anywhere under src/components/solid/ to direct DOM ops, while
// the rest of the vanilla codebase keeps building unchanged. The plugin is a
// no-op for files that don't contain Solid syntax.
import solidPlugin from "vite-plugin-solid";
const host = process.env.TAURI_DEV_HOST;
@@ -14,7 +19,13 @@ function stripCrossOrigin(): Plugin {
}
export default defineConfig({
plugins: [stripCrossOrigin()],
plugins: [
// Solid first so its JSX transform runs before any other transforms.
solidPlugin({
include: ["src/components/solid/**/*.{ts,tsx,js,jsx}"],
}),
stripCrossOrigin(),
],
build: {
modulePreload: { polyfill: false },
cssCodeSplit: false,
+274
View File
@@ -0,0 +1,274 @@
# Phase B + C — Local Follow-up TODO
This file enumerates everything from `phase-b-acceleration.md` and
`phase-c-differentiation.md` that **could not be completed inside the
sandboxed Claude session** because the work requires:
- network access to fetch new modules / npm packages,
- a Go toolchain matching `go.mod`'s `go 1.25.0` directive,
- a WASM toolchain (TinyGo / Rust / AssemblyScript),
- a real machine that can run `npm install`, `cargo`, `tauri`, etc.
The session **branch is `claude/plan-phases-b-c-bGpoS`**. Everything below
must be run on a developer machine (or CI) before the branch is mergeable.
The session-resident plan that was actually executed lives in
`/root/.claude/plans/woolly-wiggling-wolf.md` (not in this repo).
---
## Verification (do first — confirms the in-session work compiles)
- [ ] `cd Server && go build ./...` — The repo's `go.mod` requires
Go 1.25.0; the sandbox only had 1.24.7, so `go build` and `go vet`
could not be run. Manual file-by-file audit found no errors, but a
compile is the source of truth.
- [ ] `cd Server && go test ./store/... ./ws/... ./plugin/... ./telemetry/...`
— Exercises the new EventStore, EventPersister, telemetry no-op
provider, and plugin manifest/loader tests.
- [ ] `cd Server && go vet ./...`
- [ ] `cd Client/tauri-client && npm install && npm run lint && npm run build`
— Pulls in `solid-js`, `vite-plugin-solid`, and
`@solidjs/testing-library` (added to `package.json`); confirms the
Solid pipeline compiles inside the existing Vite + TS setup.
- [ ] `cd Client/tauri-client && npm run test` — runs the new
`Badge.test.tsx` smoke test.
---
## Phase B Step 6 — Solid.js migration (rest of the components)
The session landed:
- Vite + TS toolchain wiring (`vite.config.ts`, `tsconfig.json`)
- `solid-js` + `vite-plugin-solid` + `@solidjs/testing-library` in
`package.json`
- `src/lib/solidAdapter.ts` (wraps custom stores as Solid signals)
- `src/lib/solidMount.ts` (`{mount, destroy}` adapter for Solid roots)
- `src/components/solid/Badge.tsx` — first leaf
- `src/components/solid/ChannelListItem.tsx` — store-subscribed leaf
- `src/components/solid/Badge.test.tsx` — pipeline smoke test
- `src/components/solid/README.md` — migration recipe
Still TODO locally:
- [ ] Run `npm install` and verify the build passes (sandbox had no
network).
- [ ] Migrate the remaining leaf components in
`src/components/` one PR at a time, following the recipe in
`src/components/solid/README.md`. Suggested order: presence pills,
typing indicators, message attachments, voice volume meters, then
containers (channel list, member list, message list).
- [ ] Once every leaf is migrated, replace the manual `mountSolid` calls
in containers with native Solid components and delete the old
vanilla DOM utilities (`createComponent`, factory shells) referenced
from `src/components/`.
- [ ] Add a Vitest config preset under `vitest.config.ts` that pulls in
`@solidjs/testing-library` automatically (currently the test imports
it directly).
---
## Phase B Step 7 — Event persistence
The session landed:
- `Server/migrations/014_events_table.sql` (SQLite)
- `events` table appended to `Server/migrations/postgres/001_initial_schema.sql`
- `Server/db/queries/sqlite/events.sql`, `Server/db/queries/postgres/events.sql`
- `Server/db/persisted_event.go` — domain type
- `EventStore` sub-interface added to `Server/store/store.go`
- SQLite implementation in `Server/store/sqlite_events.go` (raw SQL via
`*sql.DB`, no `dbgen` dependency)
- MemStore implementation in `Server/store/memstore_events.go`
- Postgres stubs returning `ErrPostgresNotImplemented`
- `Server/ws/event_persister.go` — batched async writer
- `Server/ws/event_pruner.go` — retention pruner goroutine
- Three `replayBuf.Push` call sites in `Server/ws/hub.go` now also call
`h.persistEvent(...)`
- Tiered reconnect replay in `Server/ws/serve.go` (buffer → DB → full)
- Reconnect-tier metrics in the hub + telemetry counter
- `EventPersistenceConfig` added to `Server/config/config.go` with
defaults `{enabled: true, retention_hours: 24, batch_size: 50,
batch_flush_ms: 100, pruner_interval_minutes: 60}`
- `Server/main.go` wires the persister + pruner
- `Server/ws/event_persister_test.go` — batching, drop, drain tests
Still TODO locally:
- [ ] Run `make sqlc-generate` so `db/dbgen` and `db/pgdbgen` learn about
`events.sql`. The session used raw SQL through `*sql.DB` (matching
the existing `pgdbgen` workaround), so this is optional for SQLite
but required for the postgres backend.
- [ ] Replace the postgres EventStore stubs in `Server/store/postgres.go`
with real wrappers around the generated `pgdbgen` code (the same
mechanical work tracked in `docs/phase-a-status.md` for the other
stub methods).
- [ ] Add an integration test that pushes more than 1000 events through a
real hub with a 1000-slot buffer, disconnects at seq=500, and asserts
the DB tier returns the missing events. The session test
(`event_persister_test.go`) covers the persister in isolation but
not the buffer→DB handoff inside `handleReconnect`.
- [ ] Add a `replay_source` field to the auth_ok payload so the client
can log the tier. The hub already records the tier in metrics; the
client surface change is a separate UX call.
- [ ] Document the new `event_persistence` block in `defaultYAML` inside
`Server/config/config.go` (the struct and defaults landed; the
sample config comments did not).
---
## Phase B Step 8 — OpenTelemetry
The session landed:
- `Server/telemetry/telemetry.go` — public API + no-op provider
- `Server/telemetry/telemetry_default.go` — default-build `Init`
- `Server/telemetry/telemetry_otel.go` — wazero/postgres-style build-tag
skeleton (build with `-tags otel`); compiles only when the OTel modules
are in `go.mod` and is currently a structural placeholder
- `Server/telemetry/metrics.go``AppMetrics` bundle
- `Server/telemetry/middleware.go``HTTPMiddleware` + `PrometheusHandler`
- `Server/telemetry/telemetry_test.go`
- `Server/api/router.go` mounts `telemetry.HTTPMiddleware()`
unconditionally and the Prometheus exporter when non-nil
- `Server/main.go` calls `telemetry.Init` early and defers `Shutdown`
- `TelemetryConfig` added to `Server/config/config.go`
- Spans added to `MessageService.SendMessage`,
`PermissionService.HasChannelPerm`,
`ChannelService.ListVisibleChannels`
- Reconnect-tier counter wired into `WSReconnectTierTotal` from
`Server/ws/serve.go`
Still TODO locally:
- [ ] Add the OTel modules to `go.mod`:
```sh
cd Server
go get go.opentelemetry.io/otel@latest \
go.opentelemetry.io/otel/sdk@latest \
go.opentelemetry.io/otel/exporters/prometheus@latest \
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc@latest \
go.opentelemetry.io/contrib/instrumentation/github.com/go-chi/chi/v5/otelchi@latest
go mod tidy
```
- [ ] Replace the placeholder body of `telemetry/telemetry_otel.go`'s
`Init` with the real tracer + meter provider construction and the
`otelchi.Middleware` wiring (see the inline TODO comment with the
call graph).
- [ ] Build with `-tags otel` once the SDK is in `go.mod` and add a CI
job that exercises the tagged build.
- [ ] Add spans to the remaining service-layer entry points
(`DMService`, `VoiceService`, `InviteService`, `ModerationService`,
`BlockService`, `UserService`). The pattern is identical to the
three already done in this branch.
- [ ] Document the new `telemetry` block in `defaultYAML` inside
`Server/config/config.go`.
- [ ] Add a `make otel-up` target that spins up Jaeger via
docker-compose for local tracing development.
---
## Phase C Step 9 — Wazero plugin runtime
The session landed:
- `Server/plugin/manifest.go` — JSON manifest parser + capability checks
- `Server/plugin/loader.go` — directory scan + entrypoint validation
- `Server/plugin/registry.go` — registry + lifecycle (install/enable/uninstall)
- `Server/plugin/host_commands.go`, `host_storage.go`, `host_events.go`,
`host_http.go`, `host_ui.go` — capability surfaces
- `Server/plugin/sandbox_default.go` — no-op runtime (default build)
- `Server/plugin/sandbox_wazero.go` — `-tags wazero` skeleton
- `Server/plugin/errors.go`
- `Server/plugin/plugin_test.go`
- `Server/plugin/examples/hello/plugin.json` + `README.md`
- `Server/migrations/015_plugins.sql` (SQLite)
- `plugins` + `plugin_kv` tables appended to the postgres schema
- `Server/db/queries/sqlite/plugins.sql`,
`Server/db/queries/postgres/plugins.sql`
- `PluginStore` sub-interface in `Server/store/store.go` with SQLite,
MemStore, and postgres-stub implementations
- `Server/api/plugins_handler.go` — admin REST surface
- `Server/api/router.go` mounts the admin plugin handler
- `Server/main.go` constructs and starts the registry when
`cfg.Plugins.Enabled`
- `PluginsConfig` added to `Server/config/config.go`
- `Client/tauri-client/src/lib/pluginBridge.ts` — iframe + postMessage host
- `Client/tauri-client/src/components/solid/PluginContainer.tsx` — Solid
host component for plugin tabs
Still TODO locally:
- [ ] Add wazero to `go.mod`:
```sh
cd Server
go get github.com/tetratelabs/wazero@latest
go mod tidy
```
- [ ] Replace the placeholder body in `Server/plugin/sandbox_wazero.go`
with real wazero runtime construction. The file contains an inline
TODO with the exact API call graph.
- [ ] Replace JSON-only manifest parsing with TOML support behind the
`wazero` build tag (the design doc names `plugin.toml`). Add
`github.com/BurntSushi/toml` and a `parseTOML` shim that falls back
to the existing `ParseManifest` if no `plugin.toml` is found.
- [ ] Wire `Server/plugin/host_events.go` into the WS pub/sub hub
(`Server/ws/pubsub.go`). The session left this as a stub because
the registration surface needs to be designed alongside the actual
plugin event format — the hub-side code path is straightforward
once the format is fixed.
- [ ] Wire `Server/plugin/host_commands.go` into the WS slash-command
dispatcher. **There is currently no slash-command dispatcher in the
WS layer.** Either add one (small surface) or fold plugin commands
into the REST layer first. The plugin Registry already exposes
`DispatchCommand` so the hookup is one call site.
- [ ] Pass the live `*plugin.Registry` from `Server/main.go` into
`NewPluginAdminHandler` (the router currently constructs the
handler with a nil registry so list works but enable/disable returns
503).
- [ ] Add a precompiled trivial `.wasm` blob under
`Server/plugin/examples/hello/hello.wasm` so the example plugin can
actually be loaded by an integration test once wazero is wired.
Build it locally with TinyGo:
```sh
cd Server/plugin/examples/hello
tinygo build -o hello.wasm -target wasi ./main.go
```
- [ ] Implement plugin marketplace install path
(`POST /api/v1/admin/plugins/install` with multipart zip). The
handler is scaffolded but the install endpoint is currently absent.
- [ ] Replace plugin postgres stubs in `Server/store/postgres.go` with
real `pgdbgen`-backed implementations once `make sqlc-generate`
runs (same blocker as Phase B Step 7).
- [ ] Build the first real plugin: game detection. Pulls Steam API,
tracks playtime, exposes `/playtime` slash command. This is the
acceptance criterion in `phase-c-differentiation.md`.
---
## Build-tag matrix the user should set up in CI
| Tag set | What it builds | Why |
|---|---|---|
| (none) | Default sqlite-only server, no OTel SDK, no wazero | Existing path |
| `otel` | Above + OpenTelemetry SDK + Prometheus exporter | Phase B Step 8 |
| `wazero` | Above + plugin runtime executes WASM modules | Phase C Step 9 |
| `postgres` | Replaces sqlite with postgres backend | Phase A pending |
| `otel,wazero,postgres` | Full community-hub build | Production target |
Each tag is independently selectable; CI should test every combination at
least minimally so the build-tag boundaries don't drift.
---
## Things explicitly **out of scope** for this branch
(Documenting so reviewers don't expect them.)
- Migration of the entire vanilla TypeScript component tree to Solid. Two
proof-of-concept components landed; the rest is mechanical PRs.
- Full OTel SDK wiring (only the public API + no-op default + structural
build-tag skeleton landed).
- Real Wazero `.wasm` execution (only the registry, host APIs, and a
build-tag skeleton landed).
- Real game-detection plugin (the manifest fields and host APIs needed to
build it are in place).
- A reverse postgres → sqlite migration (Phase A documented this is
deliberately unavailable; nothing changed here).
+106
View File
@@ -0,0 +1,106 @@
// Phase C Step 9 — Plugin admin REST surface.
//
// All endpoints are mounted under the existing AdminIPRestrict group so they
// inherit the same network ACL as the rest of the admin panel. Authentication
// is handled by the admin handler's middleware before this handler runs.
package api
import (
"net/http"
"strconv"
"github.com/go-chi/chi/v5"
"github.com/owncord/server/plugin"
"github.com/owncord/server/store"
)
// PluginAdminHandler exposes plugin lifecycle operations to the admin panel.
type PluginAdminHandler struct {
registry *plugin.Registry
store store.PluginStore
}
// NewPluginAdminHandler builds an http.Handler that the router can mount.
// Pass a nil registry when plugin support is disabled — the handler then
// reports an empty list and 503 on lifecycle calls.
func NewPluginAdminHandler(registry *plugin.Registry, st store.PluginStore) http.Handler {
h := &PluginAdminHandler{registry: registry, store: st}
r := chi.NewRouter()
r.Get("/", h.list)
r.Post("/{id}/enable", h.enable)
r.Post("/{id}/disable", h.disable)
r.Delete("/{id}", h.uninstall)
return r
}
func (h *PluginAdminHandler) list(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
if h.store == nil {
writeJSON(w, http.StatusOK, []any{})
return
}
rows, err := h.store.ListPlugins(ctx)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
writeJSON(w, http.StatusOK, rows)
}
func (h *PluginAdminHandler) enable(w http.ResponseWriter, r *http.Request) {
id, ok := parsePluginID(w, r)
if !ok {
return
}
if h.registry == nil {
http.Error(w, "plugin runtime disabled", http.StatusServiceUnavailable)
return
}
if err := h.registry.EnablePlugin(r.Context(), id); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
func (h *PluginAdminHandler) disable(w http.ResponseWriter, r *http.Request) {
id, ok := parsePluginID(w, r)
if !ok {
return
}
if h.registry == nil {
http.Error(w, "plugin runtime disabled", http.StatusServiceUnavailable)
return
}
if err := h.registry.DisablePlugin(r.Context(), id); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
func (h *PluginAdminHandler) uninstall(w http.ResponseWriter, r *http.Request) {
id, ok := parsePluginID(w, r)
if !ok {
return
}
if h.registry == nil {
http.Error(w, "plugin runtime disabled", http.StatusServiceUnavailable)
return
}
if err := h.registry.UninstallPlugin(r.Context(), id); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
func parsePluginID(w http.ResponseWriter, r *http.Request) (int64, bool) {
idStr := chi.URLParam(r, "id")
id, err := strconv.ParseInt(idStr, 10, 64)
if err != nil || id <= 0 {
http.Error(w, "invalid plugin id", http.StatusBadRequest)
return 0, false
}
return id, true
}
+20
View File
@@ -19,6 +19,7 @@ import (
"github.com/owncord/server/service"
"github.com/owncord/server/storage"
dbstore "github.com/owncord/server/store"
"github.com/owncord/server/telemetry"
"github.com/owncord/server/updater"
"github.com/owncord/server/ws"
)
@@ -37,6 +38,10 @@ func NewRouter(cfg *config.Config, database *db.DB, ver string, logBuf *admin.Ri
// handled explicitly in clientIPWithProxies using the trusted_proxies config.
r.Use(middleware.Recoverer)
r.Use(requestLogger) // structured request/response logging
// Phase B Step 8 — OpenTelemetry HTTP tracing. No-op when telemetry is
// disabled or the otel build tag is not set, so this is safe to mount
// unconditionally.
r.Use(telemetry.HTTPMiddleware())
r.Use(SecurityHeadersWithTLS(cfg.TLS.Mode))
r.Use(MaxBodySizeUnless(defaultMaxBodySize, "/api/v1/uploads")) // upload route exempt
@@ -203,6 +208,15 @@ func NewRouter(cfg *config.Config, database *db.DB, ver string, logBuf *admin.Ri
func(ctx context.Context) (bool, error) { return hub.LiveKitHealthCheck(ctx) },
))
// Phase B Step 8 — OpenTelemetry Prometheus exporter. Mounted alongside
// the legacy JSON endpoint when a Prometheus exporter is wired (otel
// build, exporter == "prometheus"). Returns 404 in the default no-op build
// because telemetry.PrometheusHandler() returns nil.
if promH := telemetry.PrometheusHandler(); promH != nil {
r.With(AdminIPRestrict(cfg.Server.AdminAllowedCIDRs, cfg.Server.TrustedProxies)).
Mount("/metrics", promH)
}
// Admin panel: static files + REST API (Phase 6).
// Restrict /admin to configured CIDRs (default: private networks only).
u := updater.NewUpdater(ver, cfg.GitHub.Token, "J3vb", "OwnCord")
@@ -210,6 +224,12 @@ func NewRouter(cfg *config.Config, database *db.DB, ver string, logBuf *admin.Ri
r.Group(func(r chi.Router) {
r.Use(AdminIPRestrict(cfg.Server.AdminAllowedCIDRs, cfg.Server.TrustedProxies))
r.Mount("/admin", adminHandler)
// Phase C Step 9 — plugin admin REST surface. Mounted alongside the
// admin panel so it inherits the same network ACL. Plugin runtime is
// owned by main.go; the handler accepts a nil registry and reports
// 503 on lifecycle calls when plugin support is disabled.
r.Mount("/api/v1/admin/plugins", NewPluginAdminHandler(nil, st))
})
// Client auto-update endpoint (unauthenticated).
+72 -6
View File
@@ -19,12 +19,59 @@ import (
// Config holds the full server configuration.
type Config struct {
Server ServerConfig `koanf:"server"`
Database DatabaseConfig `koanf:"database"`
TLS TLSConfig `koanf:"tls"`
Upload UploadConfig `koanf:"upload"`
Voice VoiceConfig `koanf:"voice"`
GitHub GitHubConfig `koanf:"github"`
Server ServerConfig `koanf:"server"`
Database DatabaseConfig `koanf:"database"`
TLS TLSConfig `koanf:"tls"`
Upload UploadConfig `koanf:"upload"`
Voice VoiceConfig `koanf:"voice"`
GitHub GitHubConfig `koanf:"github"`
EventPersistence EventPersistenceConfig `koanf:"event_persistence"`
Telemetry TelemetryConfig `koanf:"telemetry"`
Plugins PluginsConfig `koanf:"plugins"`
}
// EventPersistenceConfig (Phase B Step 7) controls the tiered event log used
// for WebSocket reconnection replay.
type EventPersistenceConfig struct {
// Enabled toggles cold-storage persistence. When false the server falls
// back to ring-buffer-only behaviour (Phase A semantics).
Enabled bool `koanf:"enabled"`
// RetentionHours is how long persisted events are kept before pruning.
RetentionHours int `koanf:"retention_hours"`
// BatchSize is the maximum number of events per persister flush.
BatchSize int `koanf:"batch_size"`
// BatchFlushMs is the maximum delay between persister flushes.
BatchFlushMs int `koanf:"batch_flush_ms"`
// PrunerIntervalMinutes is how often the pruner goroutine wakes up.
PrunerIntervalMinutes int `koanf:"pruner_interval_minutes"`
}
// TelemetryConfig (Phase B Step 8) controls the OpenTelemetry exporter.
type TelemetryConfig struct {
// Enabled toggles the OTel SDK. When false the server uses no-op
// tracer/meter providers and the legacy /metrics endpoint stays the
// only metrics surface.
Enabled bool `koanf:"enabled"`
// Exporter is "none" | "prometheus" | "otlp".
Exporter string `koanf:"exporter"`
// OTLPEndpoint is the gRPC endpoint when Exporter == "otlp".
OTLPEndpoint string `koanf:"otlp_endpoint"`
// ServiceName is the resource service.name attribute.
ServiceName string `koanf:"service_name"`
}
// PluginsConfig (Phase C Step 9) controls the Wazero plugin runtime.
type PluginsConfig struct {
// Enabled toggles plugin loading at startup.
Enabled bool `koanf:"enabled"`
// Directory is the on-disk directory scanned for plugin packages.
Directory string `koanf:"directory"`
// MaxMemoryMB caps a single plugin's WASM linear memory.
MaxMemoryMB int `koanf:"max_memory_mb"`
// CPUBudgetMs caps a single plugin invocation's CPU time.
CPUBudgetMs int `koanf:"cpu_budget_ms"`
// HTTPAllowlist enumerates host suffixes plugins may reach via host_http.
HTTPAllowlist []string `koanf:"http_allowlist"`
}
// GitHubConfig holds GitHub API settings for update checking.
@@ -138,6 +185,25 @@ func defaults() Config {
Quality: "medium",
},
GitHub: GitHubConfig{},
EventPersistence: EventPersistenceConfig{
Enabled: true,
RetentionHours: 24,
BatchSize: 50,
BatchFlushMs: 100,
PrunerIntervalMinutes: 60,
},
Telemetry: TelemetryConfig{
Enabled: false,
Exporter: "none",
ServiceName: "owncord-server",
},
Plugins: PluginsConfig{
Enabled: false,
Directory: "data/plugins",
MaxMemoryMB: 64,
CPUBudgetMs: 100,
HTTPAllowlist: []string{},
},
}
}
+27
View File
@@ -0,0 +1,27 @@
package db
import "time"
// PersistedEvent is a single broadcast event written to the events table for
// cold-replay during reconnection. The event payload is the same wire-format
// JSON the WebSocket clients receive at broadcast time, including the seq
// field injected by the hub.
//
// Phase B Step 7 (event persistence layer).
type PersistedEvent struct {
Seq int64
EventType string
ChannelID int64
Payload []byte
CreatedAt time.Time
}
// PluginRow represents a row in the plugins table (Phase C Step 9).
type PluginRow struct {
ID int64
Name string
Version string
Enabled bool
ManifestJSON string
InstalledAt time.Time
}
+14
View File
@@ -0,0 +1,14 @@
-- name: PersistEvent :one
INSERT INTO events (event_type, channel_id, payload)
VALUES ($1, $2, $3)
RETURNING seq;
-- name: GetEventsSince :many
SELECT seq, event_type, channel_id, payload, created_at
FROM events
WHERE seq > $1
ORDER BY seq ASC
LIMIT $2;
-- name: PruneEventsOlderThan :execrows
DELETE FROM events WHERE created_at < $1;
+36
View File
@@ -0,0 +1,36 @@
-- name: InstallPlugin :one
INSERT INTO plugins (name, version, manifest_json)
VALUES ($1, $2, $3)
ON CONFLICT (name) DO UPDATE
SET version = excluded.version,
manifest_json = excluded.manifest_json
RETURNING id;
-- name: EnablePlugin :exec
UPDATE plugins SET enabled = TRUE WHERE id = $1;
-- name: DisablePlugin :exec
UPDATE plugins SET enabled = FALSE WHERE id = $1;
-- name: UninstallPlugin :exec
DELETE FROM plugins WHERE id = $1;
-- name: GetPlugin :one
SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins WHERE id = $1;
-- name: GetPluginByName :one
SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins WHERE name = $1;
-- name: ListPlugins :many
SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins ORDER BY name;
-- name: PluginKVGet :one
SELECT value FROM plugin_kv WHERE plugin_id = $1 AND key = $2;
-- name: PluginKVSet :exec
INSERT INTO plugin_kv (plugin_id, key, value)
VALUES ($1, $2, $3)
ON CONFLICT (plugin_id, key) DO UPDATE SET value = excluded.value;
-- name: PluginKVDelete :exec
DELETE FROM plugin_kv WHERE plugin_id = $1 AND key = $2;
+12
View File
@@ -0,0 +1,12 @@
-- name: PersistEvent :execresult
INSERT INTO events (event_type, channel_id, payload) VALUES (?, ?, ?);
-- name: GetEventsSince :many
SELECT seq, event_type, channel_id, payload, created_at
FROM events
WHERE seq > ?
ORDER BY seq ASC
LIMIT ?;
-- name: PruneEventsOlderThan :execrows
DELETE FROM events WHERE created_at < ?;
+31
View File
@@ -0,0 +1,31 @@
-- name: InstallPlugin :execresult
INSERT INTO plugins (name, version, manifest_json) VALUES (?, ?, ?)
ON CONFLICT(name) DO UPDATE SET version = excluded.version, manifest_json = excluded.manifest_json;
-- name: EnablePlugin :exec
UPDATE plugins SET enabled = 1 WHERE id = ?;
-- name: DisablePlugin :exec
UPDATE plugins SET enabled = 0 WHERE id = ?;
-- name: UninstallPlugin :exec
DELETE FROM plugins WHERE id = ?;
-- name: GetPlugin :one
SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins WHERE id = ?;
-- name: GetPluginByName :one
SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins WHERE name = ?;
-- name: ListPlugins :many
SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins ORDER BY name;
-- name: PluginKVGet :one
SELECT value FROM plugin_kv WHERE plugin_id = ? AND key = ?;
-- name: PluginKVSet :exec
INSERT INTO plugin_kv (plugin_id, key, value) VALUES (?, ?, ?)
ON CONFLICT(plugin_id, key) DO UPDATE SET value = excluded.value;
-- name: PluginKVDelete :exec
DELETE FROM plugin_kv WHERE plugin_id = ? AND key = ?;
+69
View File
@@ -23,7 +23,11 @@ import (
"github.com/owncord/server/auth"
"github.com/owncord/server/config"
"github.com/owncord/server/db"
"github.com/owncord/server/plugin"
"github.com/owncord/server/storage"
"github.com/owncord/server/store"
"github.com/owncord/server/telemetry"
"github.com/owncord/server/ws"
)
// version is overridden at build time via -ldflags "-X main.version=1.0.0".
@@ -132,10 +136,75 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer) error {
log.Info("cleared stale voice states")
}
// ── 4b. Telemetry (Phase B Step 8) ─────────────────────────────────────
telemetryShutdown, telErr := telemetry.Init(context.Background(), cfg.Telemetry)
if telErr != nil {
log.Warn("telemetry init failed; continuing without OpenTelemetry", "error", telErr)
}
defer func() {
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := telemetryShutdown(shutdownCtx); err != nil {
log.Warn("telemetry shutdown returned error", "error", err)
}
}()
// ── 5. Build HTTP router ───────────────────────────────────────────────
router, hub, routerCleanup := api.NewRouter(cfg, database, version, logBuf)
defer routerCleanup()
// ── 5b. Wire event persistence (Phase B Step 7) ────────────────────────
// Construct a Store wrapper for the cold-tier event log + plugin KV. The
// store-everywhere refactor (Phase A pending TODO) will eventually thread
// this through NewRouter directly; for now we attach it after the fact so
// the router signature stays unchanged.
storeWrapper := store.NewSQLiteStore(database)
if cfg.EventPersistence.Enabled && hub != nil {
persister := ws.NewEventPersister(
storeWrapper,
4096,
cfg.EventPersistence.BatchSize,
time.Duration(cfg.EventPersistence.BatchFlushMs)*time.Millisecond,
)
persister.Start(context.Background())
hub.SetEventPersister(persister)
hub.SetEventStore(storeWrapper)
retention := time.Duration(cfg.EventPersistence.RetentionHours) * time.Hour
prunerInterval := time.Duration(cfg.EventPersistence.PrunerIntervalMinutes) * time.Minute
prunerCtx, prunerCancel := context.WithCancel(context.Background())
ws.StartEventPruner(prunerCtx, storeWrapper, retention, prunerInterval)
defer func() {
prunerCancel()
stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer stopCancel()
persister.Stop(stopCtx)
}()
}
// ── 5c. Wire plugin runtime (Phase C Step 9) ───────────────────────────
if cfg.Plugins.Enabled {
registry, plugErr := plugin.NewRegistry(plugin.Config{
Directory: cfg.Plugins.Directory,
MaxMemoryMB: cfg.Plugins.MaxMemoryMB,
CPUBudgetMs: cfg.Plugins.CPUBudgetMs,
HTTPAllowlist: cfg.Plugins.HTTPAllowlist,
Store: storeWrapper,
})
if plugErr != nil {
log.Warn("plugin runtime init failed; continuing without plugins", "error", plugErr)
} else {
if err := registry.LoadAll(context.Background()); err != nil {
log.Warn("plugin loader: failed to scan directory", "error", err)
}
defer func() {
closeCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = registry.Close(closeCtx)
}()
}
}
// ── 6. Start server ────────────────────────────────────────────────────
addr := fmt.Sprintf(":%d", cfg.Server.Port)
srv := &http.Server{
+14
View File
@@ -0,0 +1,14 @@
-- Phase B Step 7: Event persistence layer.
-- Stores broadcast events for cold-replay during reconnection when the
-- in-memory ring buffer no longer covers the client's last_seq. Pruned by a
-- background goroutine after the configured retention window (default 24h).
CREATE TABLE IF NOT EXISTS events (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
event_type TEXT NOT NULL,
payload BLOB NOT NULL,
channel_id INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_events_channel_seq ON events(channel_id, seq);
CREATE INDEX IF NOT EXISTS idx_events_created_at ON events(created_at);
+18
View File
@@ -0,0 +1,18 @@
-- Phase C Step 9: Wazero plugin runtime.
-- Records installed plugins and their per-plugin KV namespace. The KV store
-- is exposed to plugins via the `storage` host-API capability.
CREATE TABLE IF NOT EXISTS plugins (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
version TEXT NOT NULL,
enabled INTEGER NOT NULL DEFAULT 0,
manifest_json TEXT NOT NULL,
installed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS plugin_kv (
plugin_id INTEGER NOT NULL REFERENCES plugins(id) ON DELETE CASCADE,
key TEXT NOT NULL,
value BLOB NOT NULL,
PRIMARY KEY (plugin_id, key)
);
@@ -302,3 +302,35 @@ CREATE TABLE IF NOT EXISTS user_blocks (
);
CREATE INDEX IF NOT EXISTS idx_user_blocks_blocked ON user_blocks(blocked_id, blocker_id);
-- ── events (Phase B Step 7: event persistence) ──────────────────────────────
-- Cold-storage replay buffer for WebSocket reconnections that fall outside the
-- in-memory ring window. Pruned by a background goroutine after the configured
-- retention window (default 24h).
CREATE TABLE IF NOT EXISTS events (
seq BIGSERIAL PRIMARY KEY,
event_type TEXT NOT NULL,
payload BYTEA NOT NULL,
channel_id BIGINT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_events_channel_seq ON events(channel_id, seq);
CREATE INDEX IF NOT EXISTS idx_events_created_at ON events(created_at);
-- ── plugins (Phase C Step 9: Wazero plugin runtime) ─────────────────────────
CREATE TABLE IF NOT EXISTS plugins (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
version TEXT NOT NULL,
enabled BOOLEAN NOT NULL DEFAULT FALSE,
manifest_json TEXT NOT NULL,
installed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS plugin_kv (
plugin_id BIGINT NOT NULL REFERENCES plugins(id) ON DELETE CASCADE,
key TEXT NOT NULL,
value BYTEA NOT NULL,
PRIMARY KEY (plugin_id, key)
);
+16
View File
@@ -0,0 +1,16 @@
package plugin
import "errors"
// ErrRuntimeUnavailable is returned when the plugin runtime cannot start
// because the wazero build tag was not enabled. Default builds surface this
// error from Registry.LoadAll so the rest of the server can keep running.
var ErrRuntimeUnavailable = errors.New("plugin runtime: wazero build tag not enabled (build with -tags wazero to load .wasm plugins)")
// ErrPluginNotFound is returned when an operation references an unknown
// plugin id or name.
var ErrPluginNotFound = errors.New("plugin not found")
// ErrCapabilityNotGranted is returned when a host API call would require a
// capability the plugin's manifest did not declare.
var ErrCapabilityNotGranted = errors.New("plugin capability not granted")
+29
View File
@@ -0,0 +1,29 @@
# hello plugin
Phase C Step 9 — proof-of-life plugin used by `Server/plugin/plugin_test.go`.
## Manifest
`plugin.json` declares the `commands`, `events`, and `storage` capabilities.
The manifest is the only file the default (no-`-tags wazero`) build needs —
the registry persists it into the plugins table without executing the .wasm.
## Building the WASM
The .wasm binary is intentionally NOT checked in. Build it locally with TinyGo
or any other WASM toolchain that emits a module exporting `command_dispatch`,
`on_event`, and `_start`:
```sh
# TinyGo example (writes hello.wasm into this directory)
tinygo build -o hello.wasm -target wasi ./main.go
```
A trivial main.go that satisfies the host API is sketched in
`Server/plugin/sandbox_wazero.go`'s docstring.
## Tests
`Server/plugin/plugin_test.go` exercises the manifest parser and the loader
against this directory. It does not require the .wasm to be present —
manifest-only validation is the default-build coverage path.
+12
View File
@@ -0,0 +1,12 @@
{
"name": "hello",
"version": "0.1.0",
"author": "OwnCord",
"description": "Trivial proof-of-life plugin: registers /hello and echoes message_send events.",
"entrypoint": "hello.wasm",
"permissions": ["commands", "events", "storage"],
"resources": {
"max_memory_mb": 16,
"cpu_budget_ms": 50
}
}
+63
View File
@@ -0,0 +1,63 @@
// Phase C Step 9 — `commands` host capability.
//
// Plugins that declare the "commands" capability register one or more slash
// commands at activation time. The WS command dispatcher (Server/ws/command.go)
// calls Registry.DispatchCommand after exhausting its built-in command table.
package plugin
import (
"context"
"fmt"
"strings"
)
// CommandResult is what a plugin returns from a command invocation.
type CommandResult struct {
// Reply is sent back to the invoking user as an ephemeral message.
Reply string
// Broadcast, when set, is also broadcast to the channel.
Broadcast string
}
// RegisterCommand binds cmd to inst. Called from the activation path in the
// wazero-tagged build once the module exports its `register_commands` table.
// Default build can call it directly from tests.
func (r *Registry) RegisterCommand(cmd string, inst *Instance) error {
cmd = strings.ToLower(strings.TrimPrefix(cmd, "/"))
if cmd == "" {
return fmt.Errorf("plugin: cannot register empty command")
}
if !inst.Manifest.HasCapability(CapCommands) {
return ErrCapabilityNotGranted
}
r.mu.Lock()
defer r.mu.Unlock()
if existing, ok := r.commands[cmd]; ok && existing != inst {
return fmt.Errorf("plugin: command %q already registered by %q", cmd, existing.Manifest.Name)
}
r.commands[cmd] = inst
return nil
}
// DispatchCommand routes a slash command to the owning plugin. Returns
// (nil, false) when no plugin owns the command, letting the WS dispatcher
// fall back to the not-found response. Returns (nil, true) when the runtime
// is unavailable so the dispatcher can show a helpful error message.
func (r *Registry) DispatchCommand(ctx context.Context, userID int64, channelID int64, cmd string, args []string) (*CommandResult, bool) {
if r == nil {
return nil, false
}
cmd = strings.ToLower(strings.TrimPrefix(cmd, "/"))
r.mu.RLock()
inst, ok := r.commands[cmd]
r.mu.RUnlock()
if !ok {
return nil, false
}
if r.runtimePlatform == nil {
return &CommandResult{
Reply: fmt.Sprintf("plugin %q owns /%s but the wazero runtime is not built (run with -tags wazero)", inst.Manifest.Name, cmd),
}, true
}
return r.invokeCommand(ctx, inst, userID, channelID, cmd, args)
}
+69
View File
@@ -0,0 +1,69 @@
// Phase C Step 9 — `events` host capability.
//
// Plugins that subscribe to events declare topic names in their manifest.
// At activation time the wazero-tagged build wires each subscription into
// the WS pub/sub hub via Hub.Subscribe; the default build records the
// subscription in-memory only.
package plugin
import (
"context"
"sync"
)
// EventSink is the channel a subscribed plugin reads from. The wazero-tagged
// build forwards each event to the plugin's `on_event` exported function.
type EventSink struct {
mu sync.Mutex
subs map[string][]*Instance
}
// NewEventSink returns a fresh sink. Used by the registry as the central
// fan-out for plugin event delivery.
func NewEventSink() *EventSink {
return &EventSink{subs: make(map[string][]*Instance)}
}
// Subscribe binds inst to topic. Multiple plugins may subscribe to the same
// topic — events fan out to every subscriber.
func (s *EventSink) Subscribe(topic string, inst *Instance) error {
if !inst.Manifest.HasCapability(CapEvents) {
return ErrCapabilityNotGranted
}
s.mu.Lock()
defer s.mu.Unlock()
s.subs[topic] = append(s.subs[topic], inst)
return nil
}
// UnsubscribeAll removes every subscription owned by inst (called on disable).
func (s *EventSink) UnsubscribeAll(inst *Instance) {
s.mu.Lock()
defer s.mu.Unlock()
for topic, list := range s.subs {
kept := list[:0]
for _, e := range list {
if e != inst {
kept = append(kept, e)
}
}
if len(kept) == 0 {
delete(s.subs, topic)
} else {
s.subs[topic] = kept
}
}
}
// Dispatch invokes every subscriber's on_event for topic. The default build
// is a no-op; the wazero-tagged build calls into the WASM module.
func (s *EventSink) Dispatch(ctx context.Context, topic string, payload []byte) {
s.mu.Lock()
subs := append([]*Instance(nil), s.subs[topic]...)
s.mu.Unlock()
for _, inst := range subs {
_ = inst // wazero-tagged build calls inst.module.invoke("on_event", payload)
_ = ctx
_ = payload
}
}
+94
View File
@@ -0,0 +1,94 @@
// Phase C Step 9 — `http` host capability.
//
// Outbound HTTP requests proxied through the server. Each request is matched
// against PluginsConfig.HTTPAllowlist (host suffix match) before being sent.
// The wazero-tagged build invokes this from the plugin's `host_http_request`
// import; the default build exposes it for testing.
package plugin
import (
"context"
"fmt"
"io"
"net/http"
"strings"
"time"
)
// HTTPRequest is the plugin → host request envelope.
type HTTPRequest struct {
Method string
URL string
Body []byte
Header map[string]string
}
// HTTPResponse is the host → plugin response envelope.
type HTTPResponse struct {
StatusCode int
Body []byte
Header map[string]string
}
const httpTimeout = 10 * time.Second
// HTTPDo executes a plugin-initiated HTTP request after enforcing the host
// allowlist declared in PluginsConfig.
func (r *Registry) HTTPDo(ctx context.Context, inst *Instance, req HTTPRequest) (*HTTPResponse, error) {
if !inst.Manifest.HasCapability(CapHTTP) {
return nil, ErrCapabilityNotGranted
}
if !r.hostAllowed(req.URL) {
return nil, fmt.Errorf("plugin http: host not in allowlist: %s", req.URL)
}
httpReq, err := http.NewRequestWithContext(ctx, req.Method, req.URL, strings.NewReader(string(req.Body)))
if err != nil {
return nil, fmt.Errorf("plugin http: build request: %w", err)
}
for k, v := range req.Header {
httpReq.Header.Set(k, v)
}
client := &http.Client{Timeout: httpTimeout}
resp, err := client.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("plugin http: do: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("plugin http: read body: %w", err)
}
hdr := make(map[string]string, len(resp.Header))
for k, v := range resp.Header {
if len(v) > 0 {
hdr[k] = v[0]
}
}
return &HTTPResponse{
StatusCode: resp.StatusCode,
Body: body,
Header: hdr,
}, nil
}
// hostAllowed reports whether url's host matches any suffix in the allowlist.
func (r *Registry) hostAllowed(url string) bool {
// Trivial host extraction — full URL parsing would be overkill since the
// allowlist match is suffix-based.
rest := url
for _, prefix := range []string{"https://", "http://"} {
if strings.HasPrefix(rest, prefix) {
rest = rest[len(prefix):]
break
}
}
if i := strings.IndexAny(rest, "/?#"); i >= 0 {
rest = rest[:i]
}
for _, suffix := range r.cfg.HTTPAllowlist {
if strings.HasSuffix(rest, suffix) {
return true
}
}
return false
}
+54
View File
@@ -0,0 +1,54 @@
// Phase C Step 9 — `storage` host capability.
//
// Plugins get a per-plugin namespaced KV store backed by the PluginStore
// rows in the events/plugin schema. Capacity caps and value-size caps are
// enforced here so a misbehaving plugin can't fill the database.
package plugin
import (
"context"
"fmt"
)
const (
maxPluginValueBytes = 64 * 1024 // 64 KB per value
maxPluginScanLimit = 1000 // hard cap on PluginKVScan results
)
// StoragePut writes a single key/value pair on behalf of inst.
func (r *Registry) StoragePut(ctx context.Context, inst *Instance, key string, value []byte) error {
if !inst.Manifest.HasCapability(CapStorage) {
return ErrCapabilityNotGranted
}
if len(value) > maxPluginValueBytes {
return fmt.Errorf("plugin storage: value exceeds %d bytes", maxPluginValueBytes)
}
return r.cfg.Store.PluginKVSet(ctx, inst.ID, key, value)
}
// StorageGet returns the value for key, or (nil, error) when missing.
func (r *Registry) StorageGet(ctx context.Context, inst *Instance, key string) ([]byte, error) {
if !inst.Manifest.HasCapability(CapStorage) {
return nil, ErrCapabilityNotGranted
}
return r.cfg.Store.PluginKVGet(ctx, inst.ID, key)
}
// StorageDelete removes a key.
func (r *Registry) StorageDelete(ctx context.Context, inst *Instance, key string) error {
if !inst.Manifest.HasCapability(CapStorage) {
return ErrCapabilityNotGranted
}
return r.cfg.Store.PluginKVDelete(ctx, inst.ID, key)
}
// StorageScan returns all keys with the given prefix, capped at maxPluginScanLimit.
func (r *Registry) StorageScan(ctx context.Context, inst *Instance, prefix string, limit int) (map[string][]byte, error) {
if !inst.Manifest.HasCapability(CapStorage) {
return nil, ErrCapabilityNotGranted
}
if limit <= 0 || limit > maxPluginScanLimit {
limit = maxPluginScanLimit
}
return r.cfg.Store.PluginKVScan(ctx, inst.ID, prefix, limit)
}
+62
View File
@@ -0,0 +1,62 @@
// Phase C Step 9 — `ui` host capability.
//
// A plugin that declares the `ui` capability ships HTML/CSS/JS assets and a
// list of tabs. The host serves those assets at /api/v1/plugins/<name>/ui/...
// and the Solid.js client bridge renders each tab inside a sandboxed iframe.
package plugin
import (
"net/http"
"path/filepath"
"strings"
)
// RegisterUI binds inst's declared tabs into the registry. Called from the
// activation path; safe to call multiple times (idempotent on inst).
func (r *Registry) RegisterUI(inst *Instance) error {
if !inst.Manifest.HasCapability(CapUI) {
return ErrCapabilityNotGranted
}
r.mu.Lock()
defer r.mu.Unlock()
// Drop any existing bindings for this instance, then re-add.
kept := r.uiTabs[:0]
for _, b := range r.uiTabs {
if b.PluginID != inst.ID {
kept = append(kept, b)
}
}
r.uiTabs = kept
for _, t := range inst.Manifest.UI.Tabs {
r.uiTabs = append(r.uiTabs, UITabBinding{
PluginID: inst.ID,
PluginName: inst.Manifest.Name,
Tab: t,
})
}
return nil
}
// AssetHandler returns an http.Handler that serves the on-disk assets for
// inst, rooted at the plugin's directory. The handler refuses path traversal
// attempts and only serves files declared by manifest tabs.
func (r *Registry) AssetHandler(inst *Instance) http.Handler {
allowed := make(map[string]bool, len(inst.Manifest.UI.Tabs))
for _, t := range inst.Manifest.UI.Tabs {
allowed[t.Asset] = true
}
pluginDir := filepath.Dir(inst.WASMPath)
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
rel := strings.TrimPrefix(req.URL.Path, "/")
if !allowed[rel] {
http.NotFound(w, req)
return
}
full := filepath.Join(pluginDir, rel)
if !strings.HasPrefix(full, pluginDir) {
http.Error(w, "forbidden", http.StatusForbidden)
return
}
http.ServeFile(w, req, full)
})
}
+85
View File
@@ -0,0 +1,85 @@
// Phase C Step 9 — On-disk plugin discovery.
//
// Each plugin lives in its own subdirectory under PluginsConfig.Directory:
//
// plugins/
// hello/
// plugin.json
// hello.wasm
// game-detection/
// plugin.json
// detector.wasm
// assets/...
//
// Loader walks the directory, parses every plugin.json, and returns a slice
// of foundPlugin records. The Registry then persists each into the store.
package plugin
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
)
type foundPlugin struct {
Manifest *Manifest
Dir string
WASMPath string
}
// scanPluginDirectory walks dir non-recursively and parses plugin.json from
// every immediate subdirectory. Errors on individual plugins are wrapped and
// returned alongside the successful entries.
func scanPluginDirectory(dir string) ([]foundPlugin, error) {
if dir == "" {
return nil, nil
}
entries, err := os.ReadDir(dir)
if err != nil {
if os.IsNotExist(err) {
// Directory absent is fine — operators may not have created it yet.
return nil, nil
}
return nil, err
}
var found []foundPlugin
for _, e := range entries {
if !e.IsDir() {
continue
}
pluginDir := filepath.Join(dir, e.Name())
manifestPath := filepath.Join(pluginDir, "plugin.json")
raw, rdErr := os.ReadFile(manifestPath)
if rdErr != nil {
if os.IsNotExist(rdErr) {
continue
}
return nil, fmt.Errorf("plugin %q: read plugin.json: %w", e.Name(), rdErr)
}
manifest, parseErr := ParseManifest(raw)
if parseErr != nil {
return nil, fmt.Errorf("plugin %q: %w", e.Name(), parseErr)
}
wasmPath := filepath.Join(pluginDir, manifest.Entrypoint)
if _, statErr := os.Stat(wasmPath); statErr != nil {
return nil, fmt.Errorf("plugin %q: missing entrypoint %s: %w", e.Name(), manifest.Entrypoint, statErr)
}
found = append(found, foundPlugin{
Manifest: manifest,
Dir: pluginDir,
WASMPath: wasmPath,
})
}
return found, nil
}
// serialize returns a canonical JSON encoding of the manifest, used as the
// manifest_json column value in the plugins table.
func (m *Manifest) serialize() (string, error) {
b, err := json.Marshal(m)
if err != nil {
return "", fmt.Errorf("manifest serialize: %w", err)
}
return string(b), nil
}
+129
View File
@@ -0,0 +1,129 @@
// Package plugin implements the OwnCord plugin runtime.
//
// Phase C Step 9 — Wazero Plugin Runtime.
//
// The package is split into:
//
// - manifest.go : declarative plugin metadata + permission checks
// - registry.go : in-memory registry + lifecycle (install/enable/load)
// - loader.go : on-disk discovery and package validation
// - sandbox.go : Wazero runtime configuration (build tag `wazero`)
// - host_*.go : capability-scoped host API surfaces
// - errors.go
//
// The default `go build ./...` ships a stub runtime that satisfies every call
// site without pulling Wazero into go.mod. To compile the real runtime:
//
// go get github.com/tetratelabs/wazero
// go build -tags wazero ./...
//
// This mirrors the postgres / otel build-tag approach used elsewhere in the
// repo so the default build stays self-contained.
package plugin
import (
"encoding/json"
"fmt"
"strings"
)
// Manifest is the parsed plugin metadata declared in plugin.json (or
// plugin.toml in the wazero-tagged build). The on-disk schema is intentionally
// flat so the default JSON parser handles it without a TOML dependency.
type Manifest struct {
Name string `json:"name"`
Version string `json:"version"`
Author string `json:"author"`
Description string `json:"description"`
Entrypoint string `json:"entrypoint"` // relative .wasm path
Permissions []string `json:"permissions"`
Resources Resources `json:"resources"`
UI UISpec `json:"ui"`
}
// Resources caps the plugin's runtime budget. Zero means "use the runtime
// default from PluginsConfig".
type Resources struct {
MaxMemoryMB int `json:"max_memory_mb"`
CPUBudgetMs int `json:"cpu_budget_ms"`
}
// UISpec describes the optional client-side rendering surface.
type UISpec struct {
Tabs []UITab `json:"tabs"`
}
// UITab is a single iframe-rendered plugin tab.
type UITab struct {
ID string `json:"id"`
Label string `json:"label"`
Asset string `json:"asset"` // relative html path
}
// Capability is a permission name a plugin may request.
type Capability string
const (
CapCommands Capability = "commands"
CapEvents Capability = "events"
CapStorage Capability = "storage"
CapHTTP Capability = "http"
CapUI Capability = "ui"
)
// validCapabilities is the closed set of capability names a manifest may
// declare. Anything else is rejected at load time.
var validCapabilities = map[Capability]bool{
CapCommands: true,
CapEvents: true,
CapStorage: true,
CapHTTP: true,
CapUI: true,
}
// ParseManifest decodes a plugin.json byte slice and validates required fields.
func ParseManifest(raw []byte) (*Manifest, error) {
var m Manifest
if err := json.Unmarshal(raw, &m); err != nil {
return nil, fmt.Errorf("plugin manifest: invalid JSON: %w", err)
}
if err := m.Validate(); err != nil {
return nil, err
}
return &m, nil
}
// Validate enforces the manifest schema.
func (m *Manifest) Validate() error {
if strings.TrimSpace(m.Name) == "" {
return fmt.Errorf("plugin manifest: name is required")
}
if strings.TrimSpace(m.Version) == "" {
return fmt.Errorf("plugin manifest: version is required")
}
if strings.TrimSpace(m.Entrypoint) == "" {
return fmt.Errorf("plugin manifest: entrypoint is required")
}
if !strings.HasSuffix(m.Entrypoint, ".wasm") {
return fmt.Errorf("plugin manifest: entrypoint %q must end in .wasm", m.Entrypoint)
}
for _, p := range m.Permissions {
if !validCapabilities[Capability(p)] {
return fmt.Errorf("plugin manifest: unknown permission %q", p)
}
}
if m.Resources.MaxMemoryMB < 0 || m.Resources.CPUBudgetMs < 0 {
return fmt.Errorf("plugin manifest: resources must be non-negative")
}
return nil
}
// HasCapability reports whether the manifest declared cap.
func (m *Manifest) HasCapability(cap Capability) bool {
for _, p := range m.Permissions {
if Capability(p) == cap {
return true
}
}
return false
}
+148
View File
@@ -0,0 +1,148 @@
// Phase C Step 9 — manifest + loader tests.
//
// These tests cover the default-build code path (no wazero). They confirm:
// - the JSON manifest parses and validates,
// - the loader walks a directory and surfaces well-formed plugins,
// - the registry persists discovered plugins into a PluginStore,
// - per-capability gating refuses calls when the manifest didn't grant them.
//
// Wazero-specific tests live in sandbox_wazero_test.go and only run with the
// `wazero` build tag.
package plugin
import (
"context"
"os"
"path/filepath"
"testing"
"github.com/owncord/server/store"
)
func TestParseManifestRoundTrip(t *testing.T) {
raw := []byte(`{
"name": "hello",
"version": "0.1.0",
"entrypoint": "hello.wasm",
"permissions": ["commands", "storage"],
"resources": {"max_memory_mb": 16, "cpu_budget_ms": 50}
}`)
m, err := ParseManifest(raw)
if err != nil {
t.Fatalf("ParseManifest: %v", err)
}
if m.Name != "hello" || m.Version != "0.1.0" {
t.Fatalf("unexpected manifest fields: %+v", m)
}
if !m.HasCapability(CapCommands) {
t.Fatal("expected commands capability")
}
if m.HasCapability(CapHTTP) {
t.Fatal("did not expect http capability")
}
}
func TestParseManifestRejectsBadEntrypoint(t *testing.T) {
cases := map[string]string{
"missing entrypoint": `{"name":"x","version":"1","entrypoint":""}`,
"non-wasm entrypoint": `{"name":"x","version":"1","entrypoint":"x.so"}`,
"unknown capability": `{"name":"x","version":"1","entrypoint":"x.wasm","permissions":["badperm"]}`,
"missing version": `{"name":"x","entrypoint":"x.wasm"}`,
"missing name": `{"version":"1","entrypoint":"x.wasm"}`,
}
for label, body := range cases {
t.Run(label, func(t *testing.T) {
if _, err := ParseManifest([]byte(body)); err == nil {
t.Fatalf("expected error for %s", label)
}
})
}
}
func TestScanPluginDirectoryHandlesMissing(t *testing.T) {
got, err := scanPluginDirectory(filepath.Join(t.TempDir(), "does-not-exist"))
if err != nil {
t.Fatalf("scanPluginDirectory: %v", err)
}
if len(got) != 0 {
t.Fatalf("expected empty result, got %d", len(got))
}
}
func TestScanPluginDirectoryParsesValidPlugin(t *testing.T) {
dir := t.TempDir()
pluginDir := filepath.Join(dir, "hello")
if err := os.MkdirAll(pluginDir, 0o755); err != nil {
t.Fatal(err)
}
manifest := `{"name":"hello","version":"0.1.0","entrypoint":"hello.wasm","permissions":["storage"]}`
if err := os.WriteFile(filepath.Join(pluginDir, "plugin.json"), []byte(manifest), 0o644); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(pluginDir, "hello.wasm"), []byte("\x00asm\x01\x00\x00\x00"), 0o644); err != nil {
t.Fatal(err)
}
got, err := scanPluginDirectory(dir)
if err != nil {
t.Fatalf("scanPluginDirectory: %v", err)
}
if len(got) != 1 || got[0].Manifest.Name != "hello" {
t.Fatalf("unexpected scan result: %+v", got)
}
}
func TestRegistryInstallFromDisk(t *testing.T) {
dir := t.TempDir()
pluginDir := filepath.Join(dir, "hello")
_ = os.MkdirAll(pluginDir, 0o755)
_ = os.WriteFile(filepath.Join(pluginDir, "plugin.json"),
[]byte(`{"name":"hello","version":"0.1.0","entrypoint":"hello.wasm","permissions":["storage"]}`),
0o644)
_ = os.WriteFile(filepath.Join(pluginDir, "hello.wasm"), []byte("\x00asm\x01\x00\x00\x00"), 0o644)
mem := store.NewMemStore()
reg, err := NewRegistry(Config{Directory: dir, Store: mem})
if err != nil {
t.Fatal(err)
}
if err := reg.LoadAll(context.Background()); err != nil {
t.Fatalf("LoadAll: %v", err)
}
rows, err := mem.ListPlugins(context.Background())
if err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0].Name != "hello" {
t.Fatalf("expected hello plugin row, got %+v", rows)
}
}
func TestStorageGatedByCapability(t *testing.T) {
mem := store.NewMemStore()
reg, err := NewRegistry(Config{Store: mem})
if err != nil {
t.Fatal(err)
}
inst := &Instance{
ID: 1,
Manifest: &Manifest{Name: "x", Permissions: []string{}},
}
if err := reg.StoragePut(context.Background(), inst, "k", []byte("v")); err == nil {
t.Fatal("expected ErrCapabilityNotGranted")
}
inst.Manifest.Permissions = []string{string(CapStorage)}
// Pre-create the plugin row so the KV foreign-key-equivalent succeeds.
if _, err := mem.InstallPlugin(context.Background(), "x", "0.1", "{}"); err != nil {
t.Fatal(err)
}
if err := reg.StoragePut(context.Background(), inst, "k", []byte("v")); err != nil {
t.Fatalf("StoragePut: %v", err)
}
got, err := reg.StorageGet(context.Background(), inst, "k")
if err != nil {
t.Fatalf("StorageGet: %v", err)
}
if string(got) != "v" {
t.Fatalf("expected v, got %q", got)
}
}
+254
View File
@@ -0,0 +1,254 @@
// Phase C Step 9 — Plugin registry, lifecycle, and host-API plumbing.
//
// The Registry is the long-lived handle the rest of the server holds onto. It
// owns the Wazero runtime (in the wazero-tagged build), the loaded plugin
// instances, and the dispatch tables for host-API capabilities (commands,
// events, storage, http, ui).
//
// In the default build the runtime is a stub: LoadAll walks the plugins
// directory and persists each manifest into the PluginStore so admins can see
// what is "installed", but the .wasm files are NOT executed. Calling
// Dispatch() in the default build returns ErrRuntimeUnavailable.
package plugin
import (
"context"
"fmt"
"log/slog"
"sync"
"github.com/owncord/server/store"
)
// Config is the runtime configuration sourced from PluginsConfig.
type Config struct {
Directory string
MaxMemoryMB int
CPUBudgetMs int
HTTPAllowlist []string
Store store.PluginStore
}
// Registry is the central plugin coordinator.
type Registry struct {
cfg Config
mu sync.RWMutex
plugins map[int64]*Instance // by plugin row id
byName map[string]*Instance // by manifest name
commands map[string]*Instance // command name → owning plugin
uiTabs []UITabBinding // declared by `ui` capability plugins
// runtimePlatform is set by the wazero-tagged build's NewRegistry to a
// concrete *wazero.Runtime. The default build leaves it nil and falls
// back to manifest-only behaviour.
runtimePlatform any
}
// Instance is a single loaded plugin.
type Instance struct {
ID int64
Manifest *Manifest
WASMPath string
Enabled bool
// module is the wazero compiled module in the wazero-tagged build, or
// nil in the default build.
module any
}
// UITabBinding is the public projection of a plugin's declared UI tab,
// served to the client bridge so it can render iframe tabs.
type UITabBinding struct {
PluginID int64
PluginName string
Tab UITab
}
// NewRegistry constructs a registry. In the default build it is a thin
// holder; the wazero-tagged build replaces this constructor with one that
// stands up a real Wazero runtime.
func NewRegistry(cfg Config) (*Registry, error) {
if cfg.Store == nil {
return nil, fmt.Errorf("plugin: NewRegistry requires a non-nil PluginStore")
}
return &Registry{
cfg: cfg,
plugins: make(map[int64]*Instance),
byName: make(map[string]*Instance),
commands: make(map[string]*Instance),
}, nil
}
// Close shuts the registry down. In the wazero-tagged build it tears the
// runtime down and frees module memory.
func (r *Registry) Close(ctx context.Context) error {
r.mu.Lock()
defer r.mu.Unlock()
for id := range r.plugins {
delete(r.plugins, id)
}
for n := range r.byName {
delete(r.byName, n)
}
for c := range r.commands {
delete(r.commands, c)
}
r.uiTabs = nil
return nil
}
// LoadAll scans cfg.Directory and persists every plugin.json found into the
// PluginStore. In the wazero-tagged build it then compiles each entrypoint
// into a runnable module; the default build stops at the persistence step.
func (r *Registry) LoadAll(ctx context.Context) error {
if r == nil {
return nil
}
manifests, err := scanPluginDirectory(r.cfg.Directory)
if err != nil {
return fmt.Errorf("plugin: scan %q: %w", r.cfg.Directory, err)
}
for _, found := range manifests {
if err := r.installFromDisk(ctx, found); err != nil {
slog.Warn("plugin: failed to install from disk", "name", found.Manifest.Name, "err", err)
continue
}
}
return r.activateAll(ctx)
}
// installFromDisk persists a manifest discovered on disk into the PluginStore
// and registers it in the in-memory registry.
func (r *Registry) installFromDisk(ctx context.Context, found foundPlugin) error {
manifestJSON, err := found.Manifest.serialize()
if err != nil {
return err
}
id, err := r.cfg.Store.InstallPlugin(ctx, found.Manifest.Name, found.Manifest.Version, manifestJSON)
if err != nil {
return fmt.Errorf("InstallPlugin: %w", err)
}
r.mu.Lock()
defer r.mu.Unlock()
inst := &Instance{
ID: id,
Manifest: found.Manifest,
WASMPath: found.WASMPath,
Enabled: false,
}
r.plugins[id] = inst
r.byName[found.Manifest.Name] = inst
return nil
}
// activateAll attempts to compile + register host-API hooks for every plugin
// row in the PluginStore that is marked enabled. The default build is a
// no-op (no Wazero modules to compile).
func (r *Registry) activateAll(ctx context.Context) error {
rows, err := r.cfg.Store.ListPlugins(ctx)
if err != nil {
return fmt.Errorf("ListPlugins: %w", err)
}
for _, row := range rows {
if !row.Enabled {
continue
}
r.mu.Lock()
inst, ok := r.byName[row.Name]
r.mu.Unlock()
if !ok {
slog.Warn("plugin: enabled row has no on-disk manifest, skipping", "name", row.Name)
continue
}
if err := r.activate(ctx, inst); err != nil {
slog.Warn("plugin: activation failed", "name", row.Name, "err", err)
}
}
return nil
}
// activate compiles and starts a single plugin module. Default build returns
// ErrRuntimeUnavailable; the wazero-tagged build replaces this with the real
// implementation via the runtimePlatform field.
func (r *Registry) activate(ctx context.Context, inst *Instance) error {
if r.runtimePlatform == nil {
return ErrRuntimeUnavailable
}
return r.activateWithRuntime(ctx, inst)
}
// EnablePlugin marks a plugin enabled in the store, then attempts to load it.
func (r *Registry) EnablePlugin(ctx context.Context, id int64) error {
if err := r.cfg.Store.EnablePlugin(ctx, id); err != nil {
return err
}
r.mu.RLock()
inst, ok := r.plugins[id]
r.mu.RUnlock()
if !ok {
return ErrPluginNotFound
}
inst.Enabled = true
if err := r.activate(ctx, inst); err != nil {
// Roll back the DB flag so the next start attempt is consistent.
_ = r.cfg.Store.DisablePlugin(ctx, id)
inst.Enabled = false
return err
}
return nil
}
// DisablePlugin marks a plugin disabled and tears its module down.
func (r *Registry) DisablePlugin(ctx context.Context, id int64) error {
if err := r.cfg.Store.DisablePlugin(ctx, id); err != nil {
return err
}
r.mu.Lock()
defer r.mu.Unlock()
if inst, ok := r.plugins[id]; ok {
inst.Enabled = false
// Drop command bindings owned by this plugin.
for cmd, owner := range r.commands {
if owner == inst {
delete(r.commands, cmd)
}
}
}
return nil
}
// UninstallPlugin removes a plugin entirely.
func (r *Registry) UninstallPlugin(ctx context.Context, id int64) error {
_ = r.DisablePlugin(ctx, id)
if err := r.cfg.Store.UninstallPlugin(ctx, id); err != nil {
return err
}
r.mu.Lock()
defer r.mu.Unlock()
if inst, ok := r.plugins[id]; ok {
delete(r.byName, inst.Manifest.Name)
}
delete(r.plugins, id)
return nil
}
// List returns the currently registered plugins. Read-only snapshot.
func (r *Registry) List() []*Instance {
r.mu.RLock()
defer r.mu.RUnlock()
out := make([]*Instance, 0, len(r.plugins))
for _, p := range r.plugins {
out = append(out, p)
}
return out
}
// UITabBindings returns the declared UI tabs across enabled plugins.
func (r *Registry) UITabBindings() []UITabBinding {
r.mu.RLock()
defer r.mu.RUnlock()
out := make([]UITabBinding, len(r.uiTabs))
copy(out, r.uiTabs)
return out
}
+24
View File
@@ -0,0 +1,24 @@
//go:build !wazero
// Default plugin runtime: no Wazero. Plugin manifests are still discovered,
// persisted, and surfaced through the admin API, but `.wasm` modules are not
// executed. To enable real WASM execution build with `-tags wazero`.
package plugin
import (
"context"
)
// activateWithRuntime is a no-op in the default build. It is only called from
// Registry.activate when runtimePlatform is non-nil, which never happens here.
func (r *Registry) activateWithRuntime(ctx context.Context, inst *Instance) error {
return ErrRuntimeUnavailable
}
// invokeCommand returns an error result instructing the operator to enable
// the wazero build tag. Default build only.
func (r *Registry) invokeCommand(ctx context.Context, inst *Instance, userID, channelID int64, cmd string, args []string) (*CommandResult, bool) {
return &CommandResult{
Reply: "plugin runtime disabled — rebuild server with -tags wazero to execute plugin commands",
}, true
}
+79
View File
@@ -0,0 +1,79 @@
//go:build wazero
// Real Wazero-backed plugin runtime. Compiled only with `-tags wazero`,
// matching the postgres / otel build-tag pattern used elsewhere in the repo.
//
// IMPORTANT: This file is a structural skeleton — it will fail to compile
// until github.com/tetratelabs/wazero is added to go.mod. To finish wiring it
// on a machine with network access:
//
// cd Server
// go get github.com/tetratelabs/wazero@latest
// go mod tidy
// go build -tags wazero ./...
//
// The skeleton documents the intended call graph so the implementation work
// is mechanical: each TODO marker maps to a wazero API call.
package plugin
import (
"context"
"fmt"
"os"
)
// activateWithRuntime compiles inst.WASMPath into a wazero module, applies
// the per-plugin resource caps, and registers exported functions for each
// declared capability.
func (r *Registry) activateWithRuntime(ctx context.Context, inst *Instance) error {
wasmBytes, err := os.ReadFile(inst.WASMPath)
if err != nil {
return fmt.Errorf("plugin %q: read wasm: %w", inst.Manifest.Name, err)
}
_ = wasmBytes
_ = ctx
// TODO(wazero): replace with the real wiring once go.mod has wazero:
//
// runtime := r.runtimePlatform.(wazero.Runtime)
// compiled, err := runtime.CompileModule(ctx, wasmBytes)
// if err != nil { return fmt.Errorf("compile: %w", err) }
//
// modCfg := wazero.NewModuleConfig().
// WithName(inst.Manifest.Name).
// WithStdout(io.Discard).
// WithStderr(io.Discard)
//
// memBytes := uint32(inst.Manifest.Resources.MaxMemoryMB)
// if memBytes == 0 { memBytes = uint32(r.cfg.MaxMemoryMB) }
// // wazero pages are 64 KiB; the runtime config caps via WithMemoryLimitPages.
//
// module, err := runtime.InstantiateModule(ctx, compiled, modCfg)
// if err != nil { return fmt.Errorf("instantiate: %w", err) }
//
// inst.module = module
//
// // Walk inst.Manifest.Permissions and call host_*.Register* for each
// // capability so the runtime knows what exports to look for.
return fmt.Errorf("plugin %q: wazero runtime skeleton incomplete (see sandbox_wazero.go)", inst.Manifest.Name)
}
// invokeCommand calls the plugin's `command_dispatch` exported function with
// the marshalled command + args, and decodes the response into a CommandResult.
func (r *Registry) invokeCommand(ctx context.Context, inst *Instance, userID, channelID int64, cmd string, args []string) (*CommandResult, bool) {
_ = ctx
_ = inst
_ = userID
_ = channelID
_ = cmd
_ = args
// TODO(wazero): real call:
// fn := module.ExportedFunction("command_dispatch")
// payload := encodeCommand(userID, channelID, cmd, args)
// result, err := fn.Call(ctx, ...)
// ...
return &CommandResult{
Reply: "plugin runtime: command dispatch not yet implemented in skeleton",
}, true
}
+13
View File
@@ -1,6 +1,7 @@
package service
import (
"context"
"fmt"
"log/slog"
"time"
@@ -8,6 +9,7 @@ import (
"github.com/owncord/server/db"
"github.com/owncord/server/permissions"
"github.com/owncord/server/store"
"github.com/owncord/server/telemetry"
)
// ChannelService handles channel-related business logic including
@@ -28,6 +30,17 @@ func NewChannelService(st store.Store, perms *PermissionService) *ChannelService
// ListVisibleChannels returns channels the user has ReadMessages permission for.
// DM channels are excluded (they are accessed via DMService).
func (s *ChannelService) ListVisibleChannels(userID int64) ([]db.Channel, error) {
// Phase B Step 8 — span the public service entrypoint.
ctx, span := telemetry.GlobalTracer("service/channel").Start(context.Background(),
"ChannelService.ListVisibleChannels",
telemetry.Int64("user_id", userID),
)
start := time.Now()
defer func() {
telemetry.TimeSince(ctx, telemetry.NewAppMetrics().ServiceCallDurationMs, start,
telemetry.String("method", "ListVisibleChannels"))
span.End()
}()
all, err := s.st.ListChannels()
if err != nil {
slog.Error("ChannelService.ListVisibleChannels", "err", err)
+15
View File
@@ -1,6 +1,7 @@
package service
import (
"context"
"errors"
"fmt"
"log/slog"
@@ -12,6 +13,7 @@ import (
"github.com/owncord/server/db"
"github.com/owncord/server/permissions"
"github.com/owncord/server/store"
"github.com/owncord/server/telemetry"
)
// sanitizer is the shared HTML sanitization policy (strips all tags).
@@ -115,6 +117,19 @@ func NewMessageService(st store.Store, perms *PermissionService, limiter *auth.R
// SendMessage validates, persists, and prepares broadcast data for a new message.
// Callers are responsible for emitting the appropriate events.
func (s *MessageService) SendMessage(p SendMessageParams) (*SendMessageResult, error) {
// Phase B Step 8 — wrap the public service entrypoint in a tracing span
// and a duration histogram. Both are no-ops in the default build.
ctx, span := telemetry.GlobalTracer("service/message").Start(context.Background(), "MessageService.SendMessage",
telemetry.Int64("user_id", p.UserID),
telemetry.Int64("channel_id", p.ChannelID),
)
start := time.Now()
defer func() {
telemetry.TimeSince(ctx, telemetry.NewAppMetrics().ServiceCallDurationMs, start,
telemetry.String("method", "SendMessage"))
span.End()
}()
// Rate limit.
ratKey := fmt.Sprintf("chat:%d", p.UserID)
if s.limiter != nil && !s.limiter.Allow(ratKey, 10, time.Second) {
+11
View File
@@ -1,12 +1,14 @@
package service
import (
"context"
"sync"
"time"
"github.com/owncord/server/db"
"github.com/owncord/server/permissions"
"github.com/owncord/server/store"
"github.com/owncord/server/telemetry"
)
// cachedPerms holds a snapshot of a user's role and channel overrides.
@@ -44,6 +46,15 @@ func NewPermissionService(st store.Store, checker *permissions.Checker) *Permiss
// HasChannelPerm reports whether the user has the required permission bits
// on the given channel. Uses cached role/override data when available.
func (s *PermissionService) HasChannelPerm(userID, channelID, perm int64) bool {
// Phase B Step 8 — span the perm check so traces show how many permission
// lookups a single REST/WS request triggers. The cache hit path is fast,
// but knowing how often it misses is the whole point of having metrics.
_, span := telemetry.GlobalTracer("service/permission").Start(context.Background(),
"PermissionService.HasChannelPerm",
telemetry.Int64("user_id", userID),
telemetry.Int64("channel_id", channelID),
)
defer span.End()
cp := s.getOrPopulate(userID)
if cp == nil {
return false
+6
View File
@@ -39,6 +39,12 @@ type MemStore struct {
blocks map[int64]map[int64]bool
// userID -> channelID -> lastReadMessageID
readStates map[int64]map[int64]int64
// Phase B Step 7 / Phase C Step 9 — events + plugin KV. Lazily initialised
// via ensureEvents() so existing tests that constructed a bare MemStore
// without these fields keep working.
eventsOnce sync.Once
eventStore *memEventStore
}
// NewMemStore creates an empty MemStore ready for use.
+279
View File
@@ -0,0 +1,279 @@
package store
import (
"context"
"fmt"
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/owncord/server/db"
)
// memEventStore is an in-memory EventStore + PluginStore implementation
// embedded into MemStore via the field below.
type memEventStore struct {
mu sync.Mutex
nextSeq atomic.Int64
events []db.PersistedEvent
plugins map[int64]*db.PluginRow
nextPID int64
pluginKV map[int64]map[string][]byte
}
func newMemEventStore() *memEventStore {
return &memEventStore{
plugins: make(map[int64]*db.PluginRow),
pluginKV: make(map[int64]map[string][]byte),
}
}
// ---------- EventStore ----------
func (m *MemStore) ensureEvents() *memEventStore {
m.eventsOnce.Do(func() {
m.eventStore = newMemEventStore()
})
return m.eventStore
}
func (m *MemStore) PersistEvent(_ context.Context, eventType string, channelID int64, payload []byte) (int64, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
seq := es.nextSeq.Add(1)
cp := make([]byte, len(payload))
copy(cp, payload)
es.events = append(es.events, db.PersistedEvent{
Seq: seq,
EventType: eventType,
ChannelID: channelID,
Payload: cp,
CreatedAt: time.Now().UTC(),
})
return seq, nil
}
func (m *MemStore) GetEventsSince(_ context.Context, afterSeq int64, limit int) ([]db.PersistedEvent, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
out := make([]db.PersistedEvent, 0)
for _, e := range es.events {
if e.Seq > afterSeq {
out = append(out, e)
if len(out) >= limit {
break
}
}
}
return out, nil
}
func (m *MemStore) GetEventsSinceForChannels(_ context.Context, afterSeq int64, channelIDs []int64, limit int) ([]db.PersistedEvent, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
allowed := make(map[int64]bool, len(channelIDs))
for _, cid := range channelIDs {
allowed[cid] = true
}
out := make([]db.PersistedEvent, 0)
for _, e := range es.events {
if e.Seq <= afterSeq {
continue
}
if e.ChannelID == 0 || allowed[e.ChannelID] {
out = append(out, e)
if len(out) >= limit {
break
}
}
}
return out, nil
}
func (m *MemStore) PruneEventsOlderThan(_ context.Context, cutoff time.Time) (int64, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
kept := es.events[:0]
var deleted int64
for _, e := range es.events {
if e.CreatedAt.Before(cutoff) {
deleted++
continue
}
kept = append(kept, e)
}
es.events = kept
return deleted, nil
}
// ---------- PluginStore ----------
func (m *MemStore) InstallPlugin(_ context.Context, name, version, manifestJSON string) (int64, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
for _, p := range es.plugins {
if p.Name == name {
p.Version = version
p.ManifestJSON = manifestJSON
return p.ID, nil
}
}
es.nextPID++
id := es.nextPID
es.plugins[id] = &db.PluginRow{
ID: id,
Name: name,
Version: version,
Enabled: false,
ManifestJSON: manifestJSON,
InstalledAt: time.Now().UTC(),
}
return id, nil
}
func (m *MemStore) EnablePlugin(_ context.Context, id int64) error {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
p, ok := es.plugins[id]
if !ok {
return fmt.Errorf("plugin %d not found", id)
}
p.Enabled = true
return nil
}
func (m *MemStore) DisablePlugin(_ context.Context, id int64) error {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
p, ok := es.plugins[id]
if !ok {
return fmt.Errorf("plugin %d not found", id)
}
p.Enabled = false
return nil
}
func (m *MemStore) UninstallPlugin(_ context.Context, id int64) error {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
delete(es.plugins, id)
delete(es.pluginKV, id)
return nil
}
func (m *MemStore) GetPlugin(_ context.Context, id int64) (*db.PluginRow, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
p, ok := es.plugins[id]
if !ok {
return nil, fmt.Errorf("plugin %d not found", id)
}
cp := *p
return &cp, nil
}
func (m *MemStore) GetPluginByName(_ context.Context, name string) (*db.PluginRow, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
for _, p := range es.plugins {
if p.Name == name {
cp := *p
return &cp, nil
}
}
return nil, fmt.Errorf("plugin %q not found", name)
}
func (m *MemStore) ListPlugins(_ context.Context) ([]db.PluginRow, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
out := make([]db.PluginRow, 0, len(es.plugins))
for _, p := range es.plugins {
out = append(out, *p)
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out, nil
}
func (m *MemStore) PluginKVGet(_ context.Context, pluginID int64, key string) ([]byte, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
bucket, ok := es.pluginKV[pluginID]
if !ok {
return nil, fmt.Errorf("kv: plugin %d not found", pluginID)
}
v, ok := bucket[key]
if !ok {
return nil, fmt.Errorf("kv: key %q not found", key)
}
cp := make([]byte, len(v))
copy(cp, v)
return cp, nil
}
func (m *MemStore) PluginKVSet(_ context.Context, pluginID int64, key string, value []byte) error {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
bucket, ok := es.pluginKV[pluginID]
if !ok {
bucket = make(map[string][]byte)
es.pluginKV[pluginID] = bucket
}
cp := make([]byte, len(value))
copy(cp, value)
bucket[key] = cp
return nil
}
func (m *MemStore) PluginKVDelete(_ context.Context, pluginID int64, key string) error {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
if bucket, ok := es.pluginKV[pluginID]; ok {
delete(bucket, key)
}
return nil
}
func (m *MemStore) PluginKVScan(_ context.Context, pluginID int64, prefix string, limit int) (map[string][]byte, error) {
es := m.ensureEvents()
es.mu.Lock()
defer es.mu.Unlock()
out := make(map[string][]byte)
bucket, ok := es.pluginKV[pluginID]
if !ok {
return out, nil
}
keys := make([]string, 0, len(bucket))
for k := range bucket {
if strings.HasPrefix(k, prefix) {
keys = append(keys, k)
}
}
sort.Strings(keys)
if limit > 0 && len(keys) > limit {
keys = keys[:limit]
}
for _, k := range keys {
v := bucket[k]
cp := make([]byte, len(v))
copy(cp, v)
out[k] = cp
}
return out, nil
}
+64
View File
@@ -574,6 +574,70 @@ func (s *PostgresStore) GetAllSettings() (map[string]string, error) {
return nil, ErrPostgresNotImplemented
}
// ── EventStore (stubs — Phase B Step 7) ─────────────────────────────────────
func (s *PostgresStore) PersistEvent(ctx context.Context, eventType string, channelID int64, payload []byte) (int64, error) {
return 0, ErrPostgresNotImplemented
}
func (s *PostgresStore) GetEventsSince(ctx context.Context, afterSeq int64, limit int) ([]db.PersistedEvent, error) {
return nil, ErrPostgresNotImplemented
}
func (s *PostgresStore) GetEventsSinceForChannels(ctx context.Context, afterSeq int64, channelIDs []int64, limit int) ([]db.PersistedEvent, error) {
return nil, ErrPostgresNotImplemented
}
func (s *PostgresStore) PruneEventsOlderThan(ctx context.Context, cutoff time.Time) (int64, error) {
return 0, ErrPostgresNotImplemented
}
// ── PluginStore (stubs — Phase C Step 9) ────────────────────────────────────
func (s *PostgresStore) InstallPlugin(ctx context.Context, name, version, manifestJSON string) (int64, error) {
return 0, ErrPostgresNotImplemented
}
func (s *PostgresStore) EnablePlugin(ctx context.Context, id int64) error {
return ErrPostgresNotImplemented
}
func (s *PostgresStore) DisablePlugin(ctx context.Context, id int64) error {
return ErrPostgresNotImplemented
}
func (s *PostgresStore) UninstallPlugin(ctx context.Context, id int64) error {
return ErrPostgresNotImplemented
}
func (s *PostgresStore) GetPlugin(ctx context.Context, id int64) (*db.PluginRow, error) {
return nil, ErrPostgresNotImplemented
}
func (s *PostgresStore) GetPluginByName(ctx context.Context, name string) (*db.PluginRow, error) {
return nil, ErrPostgresNotImplemented
}
func (s *PostgresStore) ListPlugins(ctx context.Context) ([]db.PluginRow, error) {
return nil, ErrPostgresNotImplemented
}
func (s *PostgresStore) PluginKVGet(ctx context.Context, pluginID int64, key string) ([]byte, error) {
return nil, ErrPostgresNotImplemented
}
func (s *PostgresStore) PluginKVSet(ctx context.Context, pluginID int64, key string, value []byte) error {
return ErrPostgresNotImplemented
}
func (s *PostgresStore) PluginKVDelete(ctx context.Context, pluginID int64, key string) error {
return ErrPostgresNotImplemented
}
func (s *PostgresStore) PluginKVScan(ctx context.Context, pluginID int64, prefix string, limit int) (map[string][]byte, error) {
return nil, ErrPostgresNotImplemented
}
// Compile-time interface check — fails to compile if any Store method is
// missing a PostgresStore receiver.
var _ Store = (*PostgresStore)(nil)
+290
View File
@@ -0,0 +1,290 @@
package store
import (
"context"
"fmt"
"strings"
"time"
"github.com/owncord/server/db"
)
// ── EventStore (Phase B Step 7) ─────────────────────────────────────────────
// PersistEvent appends a single event to the events table and returns the
// auto-assigned seq.
func (s *SQLiteStore) PersistEvent(ctx context.Context, eventType string, channelID int64, payload []byte) (int64, error) {
res, err := s.db.SQLDb().ExecContext(ctx,
`INSERT INTO events (event_type, channel_id, payload) VALUES (?, ?, ?)`,
eventType, channelID, payload,
)
if err != nil {
return 0, fmt.Errorf("PersistEvent: %w", err)
}
id, err := res.LastInsertId()
if err != nil {
return 0, fmt.Errorf("PersistEvent LastInsertId: %w", err)
}
return id, nil
}
// GetEventsSince returns events with seq > afterSeq up to limit, ordered ASC.
func (s *SQLiteStore) GetEventsSince(ctx context.Context, afterSeq int64, limit int) ([]db.PersistedEvent, error) {
rows, err := s.db.SQLDb().QueryContext(ctx,
`SELECT seq, event_type, channel_id, payload, created_at
FROM events
WHERE seq > ?
ORDER BY seq ASC
LIMIT ?`,
afterSeq, limit,
)
if err != nil {
return nil, fmt.Errorf("GetEventsSince: %w", err)
}
defer rows.Close()
return scanEventRows(rows)
}
// GetEventsSinceForChannels filters events to those whose channel_id is 0
// (global broadcast) or in channelIDs.
func (s *SQLiteStore) GetEventsSinceForChannels(ctx context.Context, afterSeq int64, channelIDs []int64, limit int) ([]db.PersistedEvent, error) {
// Build IN clause manually since database/sql does not expand slices.
if len(channelIDs) == 0 {
// Only global broadcasts.
rows, err := s.db.SQLDb().QueryContext(ctx,
`SELECT seq, event_type, channel_id, payload, created_at
FROM events
WHERE seq > ? AND channel_id = 0
ORDER BY seq ASC
LIMIT ?`,
afterSeq, limit,
)
if err != nil {
return nil, fmt.Errorf("GetEventsSinceForChannels (global only): %w", err)
}
defer rows.Close()
return scanEventRows(rows)
}
placeholders := make([]string, len(channelIDs))
args := make([]any, 0, len(channelIDs)+2)
args = append(args, afterSeq)
for i, cid := range channelIDs {
placeholders[i] = "?"
args = append(args, cid)
}
args = append(args, limit)
query := fmt.Sprintf(
`SELECT seq, event_type, channel_id, payload, created_at
FROM events
WHERE seq > ?
AND (channel_id = 0 OR channel_id IN (%s))
ORDER BY seq ASC
LIMIT ?`,
strings.Join(placeholders, ","),
)
rows, err := s.db.SQLDb().QueryContext(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("GetEventsSinceForChannels: %w", err)
}
defer rows.Close()
return scanEventRows(rows)
}
// PruneEventsOlderThan deletes events older than cutoff. Returns rows deleted.
func (s *SQLiteStore) PruneEventsOlderThan(ctx context.Context, cutoff time.Time) (int64, error) {
res, err := s.db.SQLDb().ExecContext(ctx,
`DELETE FROM events WHERE created_at < ?`,
cutoff.UTC().Format("2006-01-02 15:04:05"),
)
if err != nil {
return 0, fmt.Errorf("PruneEventsOlderThan: %w", err)
}
n, err := res.RowsAffected()
if err != nil {
return 0, fmt.Errorf("PruneEventsOlderThan RowsAffected: %w", err)
}
return n, nil
}
// ── PluginStore (Phase C Step 9) ────────────────────────────────────────────
func (s *SQLiteStore) InstallPlugin(ctx context.Context, name, version, manifestJSON string) (int64, error) {
res, err := s.db.SQLDb().ExecContext(ctx,
`INSERT INTO plugins (name, version, enabled, manifest_json) VALUES (?, ?, 0, ?)
ON CONFLICT(name) DO UPDATE SET version = excluded.version, manifest_json = excluded.manifest_json`,
name, version, manifestJSON,
)
if err != nil {
return 0, fmt.Errorf("InstallPlugin: %w", err)
}
id, err := res.LastInsertId()
if err != nil || id == 0 {
// On conflict path LastInsertId may be 0; look up by name.
row := s.db.SQLDb().QueryRowContext(ctx, `SELECT id FROM plugins WHERE name = ?`, name)
if scanErr := row.Scan(&id); scanErr != nil {
return 0, fmt.Errorf("InstallPlugin lookup: %w", scanErr)
}
}
return id, nil
}
func (s *SQLiteStore) EnablePlugin(ctx context.Context, id int64) error {
_, err := s.db.SQLDb().ExecContext(ctx, `UPDATE plugins SET enabled = 1 WHERE id = ?`, id)
return err
}
func (s *SQLiteStore) DisablePlugin(ctx context.Context, id int64) error {
_, err := s.db.SQLDb().ExecContext(ctx, `UPDATE plugins SET enabled = 0 WHERE id = ?`, id)
return err
}
func (s *SQLiteStore) UninstallPlugin(ctx context.Context, id int64) error {
_, err := s.db.SQLDb().ExecContext(ctx, `DELETE FROM plugins WHERE id = ?`, id)
return err
}
func (s *SQLiteStore) GetPlugin(ctx context.Context, id int64) (*db.PluginRow, error) {
row := s.db.SQLDb().QueryRowContext(ctx,
`SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins WHERE id = ?`,
id,
)
return scanPluginRow(row)
}
func (s *SQLiteStore) GetPluginByName(ctx context.Context, name string) (*db.PluginRow, error) {
row := s.db.SQLDb().QueryRowContext(ctx,
`SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins WHERE name = ?`,
name,
)
return scanPluginRow(row)
}
func (s *SQLiteStore) ListPlugins(ctx context.Context) ([]db.PluginRow, error) {
rows, err := s.db.SQLDb().QueryContext(ctx,
`SELECT id, name, version, enabled, manifest_json, installed_at FROM plugins ORDER BY name`,
)
if err != nil {
return nil, fmt.Errorf("ListPlugins: %w", err)
}
defer rows.Close()
var out []db.PluginRow
for rows.Next() {
var p db.PluginRow
var enabledInt int64
var installedAt string
if err := rows.Scan(&p.ID, &p.Name, &p.Version, &enabledInt, &p.ManifestJSON, &installedAt); err != nil {
return nil, fmt.Errorf("ListPlugins scan: %w", err)
}
p.Enabled = enabledInt != 0
p.InstalledAt = parseSQLiteTime(installedAt)
out = append(out, p)
}
return out, rows.Err()
}
func (s *SQLiteStore) PluginKVGet(ctx context.Context, pluginID int64, key string) ([]byte, error) {
row := s.db.SQLDb().QueryRowContext(ctx,
`SELECT value FROM plugin_kv WHERE plugin_id = ? AND key = ?`,
pluginID, key,
)
var v []byte
if err := row.Scan(&v); err != nil {
return nil, err
}
return v, nil
}
func (s *SQLiteStore) PluginKVSet(ctx context.Context, pluginID int64, key string, value []byte) error {
_, err := s.db.SQLDb().ExecContext(ctx,
`INSERT INTO plugin_kv (plugin_id, key, value) VALUES (?, ?, ?)
ON CONFLICT(plugin_id, key) DO UPDATE SET value = excluded.value`,
pluginID, key, value,
)
return err
}
func (s *SQLiteStore) PluginKVDelete(ctx context.Context, pluginID int64, key string) error {
_, err := s.db.SQLDb().ExecContext(ctx,
`DELETE FROM plugin_kv WHERE plugin_id = ? AND key = ?`,
pluginID, key,
)
return err
}
func (s *SQLiteStore) PluginKVScan(ctx context.Context, pluginID int64, prefix string, limit int) (map[string][]byte, error) {
rows, err := s.db.SQLDb().QueryContext(ctx,
`SELECT key, value FROM plugin_kv WHERE plugin_id = ? AND key LIKE ? ORDER BY key LIMIT ?`,
pluginID, prefix+"%", limit,
)
if err != nil {
return nil, fmt.Errorf("PluginKVScan: %w", err)
}
defer rows.Close()
out := make(map[string][]byte)
for rows.Next() {
var k string
var v []byte
if err := rows.Scan(&k, &v); err != nil {
return nil, err
}
out[k] = v
}
return out, rows.Err()
}
// ── helpers ─────────────────────────────────────────────────────────────────
type rowScanner interface {
Scan(dest ...any) error
}
func scanPluginRow(row rowScanner) (*db.PluginRow, error) {
var p db.PluginRow
var enabledInt int64
var installedAt string
if err := row.Scan(&p.ID, &p.Name, &p.Version, &enabledInt, &p.ManifestJSON, &installedAt); err != nil {
return nil, err
}
p.Enabled = enabledInt != 0
p.InstalledAt = parseSQLiteTime(installedAt)
return &p, nil
}
type rowsScanner interface {
Next() bool
Scan(dest ...any) error
Err() error
}
func scanEventRows(rows rowsScanner) ([]db.PersistedEvent, error) {
var out []db.PersistedEvent
for rows.Next() {
var e db.PersistedEvent
var createdAt string
if err := rows.Scan(&e.Seq, &e.EventType, &e.ChannelID, &e.Payload, &createdAt); err != nil {
return nil, fmt.Errorf("scanEventRows: %w", err)
}
e.CreatedAt = parseSQLiteTime(createdAt)
out = append(out, e)
}
if err := rows.Err(); err != nil {
return nil, err
}
return out, nil
}
func parseSQLiteTime(s string) time.Time {
// SQLite CURRENT_TIMESTAMP returns "YYYY-MM-DD HH:MM:SS" in UTC.
for _, layout := range []string{
"2006-01-02 15:04:05",
time.RFC3339,
time.RFC3339Nano,
} {
if t, err := time.Parse(layout, s); err == nil {
return t.UTC()
}
}
return time.Time{}
}
+43
View File
@@ -26,6 +26,8 @@ type Store interface {
AttachmentStore
AdminStore
SettingsStore
EventStore
PluginStore
// Close releases the underlying database connection.
Close() error
@@ -194,3 +196,44 @@ type SettingsStore interface {
SetSetting(key, value string) error
GetAllSettings() (map[string]string, error)
}
// EventStore persists broadcast events for cold-replay during reconnection
// when the in-memory ring buffer no longer covers the client's last_seq.
//
// Phase B Step 7 — Event Persistence Layer.
type EventStore interface {
// PersistEvent appends an event to the persistent log and returns the
// auto-assigned seq. channelID == 0 means the event was a global broadcast.
PersistEvent(ctx context.Context, eventType string, channelID int64, payload []byte) (int64, error)
// GetEventsSince returns up to limit events with seq > afterSeq, ordered
// by seq ascending. Used as a fallback after the ring buffer misses.
GetEventsSince(ctx context.Context, afterSeq int64, limit int) ([]db.PersistedEvent, error)
// GetEventsSinceForChannels returns up to limit events with seq > afterSeq
// whose channel_id is in channelIDs OR is 0 (global broadcasts), ordered by
// seq ascending. Mirrors EventRingBuffer.EventsSinceFiltered.
GetEventsSinceForChannels(ctx context.Context, afterSeq int64, channelIDs []int64, limit int) ([]db.PersistedEvent, error)
// PruneEventsOlderThan deletes events with created_at < cutoff. Returns the
// number of deleted rows. Called periodically by the pruner goroutine.
PruneEventsOlderThan(ctx context.Context, cutoff time.Time) (int64, error)
}
// PluginStore manages installed plugins and per-plugin KV namespaces.
//
// Phase C Step 9 — Wazero Plugin Runtime.
type PluginStore interface {
InstallPlugin(ctx context.Context, name, version, manifestJSON string) (int64, error)
EnablePlugin(ctx context.Context, id int64) error
DisablePlugin(ctx context.Context, id int64) error
UninstallPlugin(ctx context.Context, id int64) error
GetPlugin(ctx context.Context, id int64) (*db.PluginRow, error)
GetPluginByName(ctx context.Context, name string) (*db.PluginRow, error)
ListPlugins(ctx context.Context) ([]db.PluginRow, error)
PluginKVGet(ctx context.Context, pluginID int64, key string) ([]byte, error)
PluginKVSet(ctx context.Context, pluginID int64, key string, value []byte) error
PluginKVDelete(ctx context.Context, pluginID int64, key string) error
PluginKVScan(ctx context.Context, pluginID int64, prefix string, limit int) (map[string][]byte, error)
}
+64
View File
@@ -0,0 +1,64 @@
// Phase B Step 8 — declared application metrics.
//
// Instruments are constructed lazily against the global Provider so callers
// don't need to thread a Meter through every constructor. Hot-path callers
// should cache the returned instrument in a struct field rather than calling
// these helpers per request — they take a sync.RWMutex to read the global
// provider and the cost adds up at high throughput.
package telemetry
import "sync"
const (
scopeWS = "github.com/owncord/server/ws"
scopeService = "github.com/owncord/server/service"
scopeDB = "github.com/owncord/server/db"
scopeVoice = "github.com/owncord/server/voice"
)
// AppMetrics is the canonical bundle of meters used across the server. Build
// it once at startup with NewAppMetrics() and stash it on the relevant
// long-lived structs (Hub, services, etc).
type AppMetrics struct {
WSMessagesTotal Counter
WSActiveConnections Gauge
WSBroadcastLatency Histogram
WSReconnectTierTotal Counter
WSEventsPersisted Counter
WSEventsDropped Counter
DBQueryDurationSec Histogram
VoiceActiveSessions Gauge
VoiceParticipants Gauge
ServiceCallDurationMs Histogram
}
var (
appMetricsOnce sync.Once
appMetricsInst *AppMetrics
)
// NewAppMetrics returns a process-wide AppMetrics, lazily constructed against
// the current global provider. Calling it multiple times returns the same
// instance — the metrics are tied to the global provider, not to a specific
// caller.
func NewAppMetrics() *AppMetrics {
appMetricsOnce.Do(func() {
ws := GlobalMeter(scopeWS)
svc := GlobalMeter(scopeService)
db := GlobalMeter(scopeDB)
voice := GlobalMeter(scopeVoice)
appMetricsInst = &AppMetrics{
WSMessagesTotal: ws.Counter("ws_messages_total", "WebSocket messages broadcast"),
WSActiveConnections: ws.Gauge("ws_active_connections", "Currently connected WebSocket clients"),
WSBroadcastLatency: ws.Histogram("ws_broadcast_latency_seconds", "Wall-clock seconds from enqueue to fanout completion", "s"),
WSReconnectTierTotal: ws.Counter("ws_reconnect_tier_total", "Reconnection replay tier hits, attribute tier=buffer|db|full"),
WSEventsPersisted: ws.Counter("ws_events_persisted_total", "Events written to the cold-tier event log"),
WSEventsDropped: ws.Counter("ws_events_dropped_total", "Events dropped because the persister queue was full"),
DBQueryDurationSec: db.Histogram("db_query_duration_seconds", "Per-query wall time", "s"),
VoiceActiveSessions: voice.Gauge("voice_active_sessions", "Active LiveKit rooms"),
VoiceParticipants: voice.Gauge("voice_participants", "Connected LiveKit participants across all rooms"),
ServiceCallDurationMs: svc.Histogram("service_call_duration_seconds", "Service-layer method execution time", "s"),
}
})
return appMetricsInst
}
+21
View File
@@ -0,0 +1,21 @@
// Phase B Step 8 — HTTP middleware that delegates tracing to the global
// provider. The default no-op build returns next unchanged; the otel-tagged
// build wraps next with otelchi.Middleware. Mount it from the Chi router so
// every REST request becomes a span automatically.
package telemetry
import "net/http"
// HTTPMiddleware returns an HTTP middleware that traces every request via
// the currently installed Provider. Safe to mount unconditionally — it is a
// pass-through when telemetry is disabled.
func HTTPMiddleware() func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler {
return Global().HTTPMiddleware(next)
}
}
// PrometheusHandler returns the active Prometheus exporter handler, or nil if
// no Prometheus exporter is wired. Mount it from the API router only when
// non-nil so the legacy /metrics endpoint stays untouched in the no-op build.
func PrometheusHandler() http.Handler { return Global().PrometheusHandler() }
+172
View File
@@ -0,0 +1,172 @@
// Package telemetry wires the OpenTelemetry SDK for OwnCord.
//
// Phase B Step 8 — Add OpenTelemetry for Observability.
//
// This file provides the public API surface and a no-op default implementation
// that compiles without any external dependencies. The real OpenTelemetry
// providers live in telemetry_otel.go behind the `otel` build tag, so the
// default build (`go build ./...`) ships a stub that satisfies every call site
// at zero binary cost. To enable real OTel:
//
// go get go.opentelemetry.io/otel \
// go.opentelemetry.io/otel/sdk \
// go.opentelemetry.io/otel/exporters/prometheus \
// go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc \
// go.opentelemetry.io/contrib/instrumentation/github.com/go-chi/chi/otelchi
// go build -tags otel ./...
//
// The fallback API is intentionally tiny: it lets the rest of the codebase
// reference Tracer / Meter / Counter / Histogram without caring whether the
// real SDK is compiled in.
package telemetry
import (
"context"
"net/http"
"sync"
"time"
)
// ShutdownFunc is returned by Init and must be called on server shutdown to
// flush exporters. The no-op implementation returns nil.
type ShutdownFunc func(context.Context) error
// Counter is a monotonically increasing integer metric.
type Counter interface {
Add(ctx context.Context, delta int64, attrs ...Attr)
}
// Histogram records distributions of float64 observations (e.g. durations).
type Histogram interface {
Record(ctx context.Context, value float64, attrs ...Attr)
}
// Gauge records the current value of a measurement.
type Gauge interface {
Set(ctx context.Context, value float64, attrs ...Attr)
}
// Attr is a single key/value attribute attached to a metric or span.
type Attr struct {
Key string
Value any
}
// String constructs a string attribute.
func String(k, v string) Attr { return Attr{Key: k, Value: v} }
// Int64 constructs an int64 attribute.
func Int64(k string, v int64) Attr { return Attr{Key: k, Value: v} }
// Float64 constructs a float64 attribute.
func Float64(k string, v float64) Attr { return Attr{Key: k, Value: v} }
// Span is a single tracing span.
type Span interface {
End()
SetAttributes(attrs ...Attr)
RecordError(err error)
}
// Tracer creates spans within a single instrumentation library.
type Tracer interface {
Start(ctx context.Context, name string, attrs ...Attr) (context.Context, Span)
}
// Meter creates instruments within a single instrumentation library.
type Meter interface {
Counter(name, description string) Counter
Histogram(name, description, unit string) Histogram
Gauge(name, description string) Gauge
}
// Provider is the top-level handle returned by Init.
type Provider interface {
Tracer(name string) Tracer
Meter(name string) Meter
HTTPMiddleware(next http.Handler) http.Handler
PrometheusHandler() http.Handler // returns nil when no Prometheus exporter is wired
}
var (
globalMu sync.RWMutex
globalProvider Provider = noopProvider{}
)
// SetGlobal stores p as the package-level provider returned by Global.
func SetGlobal(p Provider) {
globalMu.Lock()
defer globalMu.Unlock()
if p == nil {
globalProvider = noopProvider{}
return
}
globalProvider = p
}
// Global returns the currently installed provider, or a no-op when none.
func Global() Provider {
globalMu.RLock()
defer globalMu.RUnlock()
return globalProvider
}
// Tracer is a convenience that fetches a tracer from the global provider.
func GlobalTracer(name string) Tracer { return Global().Tracer(name) }
// Meter is a convenience that fetches a meter from the global provider.
func GlobalMeter(name string) Meter { return Global().Meter(name) }
// ── No-op implementation ────────────────────────────────────────────────────
type noopProvider struct{}
func (noopProvider) Tracer(string) Tracer { return noopTracer{} }
func (noopProvider) Meter(string) Meter { return noopMeter{} }
func (noopProvider) HTTPMiddleware(next http.Handler) http.Handler { return next }
func (noopProvider) PrometheusHandler() http.Handler { return nil }
type noopTracer struct{}
func (noopTracer) Start(ctx context.Context, _ string, _ ...Attr) (context.Context, Span) {
return ctx, noopSpan{}
}
type noopSpan struct{}
func (noopSpan) End() {}
func (noopSpan) SetAttributes(...Attr) {}
func (noopSpan) RecordError(error) {}
type noopMeter struct{}
func (noopMeter) Counter(string, string) Counter { return noopCounter{} }
func (noopMeter) Histogram(string, string, string) Histogram { return noopHistogram{} }
func (noopMeter) Gauge(string, string) Gauge { return noopGauge{} }
type noopCounter struct{}
func (noopCounter) Add(context.Context, int64, ...Attr) {}
type noopHistogram struct{}
func (noopHistogram) Record(context.Context, float64, ...Attr) {}
type noopGauge struct{}
func (noopGauge) Set(context.Context, float64, ...Attr) {}
// TimeSince records the elapsed time since start as seconds on h. Convenience
// shim used by Hub / service-layer instrumentation.
func TimeSince(ctx context.Context, h Histogram, start time.Time, attrs ...Attr) {
if h == nil {
return
}
h.Record(ctx, time.Since(start).Seconds(), attrs...)
}
// init installs a no-op provider as the package default. The real
// telemetry_otel.go (build tag `otel`) overrides this via Init.
func init() {
SetGlobal(noopProvider{})
}
+21
View File
@@ -0,0 +1,21 @@
//go:build !otel
// Default no-op build of the telemetry package — see telemetry.go for the
// public API. The real OpenTelemetry SDK wiring lives in telemetry_otel.go and
// is selected by `go build -tags otel ./...`.
package telemetry
import (
"context"
"github.com/owncord/server/config"
)
// Init configures and installs the OpenTelemetry SDK based on cfg. In the
// default build this installs a no-op provider so call sites have a usable
// global handle without pulling in any external dependencies.
func Init(_ context.Context, cfg config.TelemetryConfig) (ShutdownFunc, error) {
_ = cfg // honoured by the otel-tagged build only
SetGlobal(noopProvider{})
return func(context.Context) error { return nil }, nil
}
+75
View File
@@ -0,0 +1,75 @@
//go:build otel
// Real OpenTelemetry-backed implementation. Compiled only with `-tags otel`,
// which keeps the OTel SDK out of the default sqlite-only build (matching the
// pattern used by Server/store/postgres.go).
//
// IMPORTANT: This file currently contains a real-API skeleton that will fail
// to compile until the OTel modules are added to go.mod. To finish wiring it,
// run on a machine with network access:
//
// cd Server
// go get go.opentelemetry.io/otel@latest \
// go.opentelemetry.io/otel/sdk@latest \
// go.opentelemetry.io/otel/exporters/prometheus@latest \
// go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc@latest \
// go.opentelemetry.io/contrib/instrumentation/github.com/go-chi/chi/v5/otelchi@latest
// go mod tidy
// go build -tags otel ./...
//
// Until that runs, the default build (no `-tags otel`) uses telemetry_default.go.
package telemetry
import (
"context"
"fmt"
"net/http"
"github.com/owncord/server/config"
)
// otelProvider is the placeholder for the real OTel-backed Provider. The
// fields and methods will be filled in once the OTel modules are in go.mod;
// for now this file exists so reviewers can see the intended shape and the
// `otel` build tag has a target.
type otelProvider struct {
cfg config.TelemetryConfig
promHandler http.Handler
httpMiddleware func(http.Handler) http.Handler
shutdown ShutdownFunc
}
// Init wires the OTel SDK exporters according to cfg.Exporter:
//
// "none" — no-op (matches the default build)
// "prometheus" — pull-based Prometheus exporter mounted at /metrics
// "otlp" — push-based OTLP/gRPC exporter to cfg.OTLPEndpoint
//
// All exporters share the same resource (service.name = cfg.ServiceName).
func Init(ctx context.Context, cfg config.TelemetryConfig) (ShutdownFunc, error) {
if !cfg.Enabled || cfg.Exporter == "" || cfg.Exporter == "none" {
SetGlobal(noopProvider{})
return func(context.Context) error { return nil }, nil
}
// TODO(otel-build-tag): replace the panic below with the real OTel
// initialisation once go.mod has the otel modules. The structural call
// graph is:
//
// resource = sdkresource.NewWithAttributes(...)
// tp = sdktrace.NewTracerProvider(WithBatcher(otlptracegrpc...))
// mp = sdkmetric.NewMeterProvider(WithReader(prometheus.New()))
// otel.SetTracerProvider(tp); otel.SetMeterProvider(mp)
// handler = promhttp.HandlerFor(prometheusReg, promhttp.HandlerOpts{})
// mw = otelchi.Middleware(serviceName, otelchi.WithChiRoutes(...))
//
// then wrap them in a Provider implementation and SetGlobal it.
_ = ctx
return nil, fmt.Errorf("telemetry: otel build tag is set but the SDK skeleton in telemetry_otel.go is incomplete; finish wiring after `go get go.opentelemetry.io/otel...`")
}
// otelProvider satisfies Provider once the SDK is wired.
func (p *otelProvider) Tracer(name string) Tracer { _ = name; return noopTracer{} }
func (p *otelProvider) Meter(name string) Meter { _ = name; return noopMeter{} }
func (p *otelProvider) HTTPMiddleware(next http.Handler) http.Handler { return p.httpMiddleware(next) }
func (p *otelProvider) PrometheusHandler() http.Handler { return p.promHandler }
+67
View File
@@ -0,0 +1,67 @@
// Phase B Step 8 — telemetry default-build smoke test.
//
// In the default build (no -tags otel) the package installs a no-op provider
// that satisfies every API surface. The test confirms Init returns a non-nil
// shutdown closer, the Global() helper returns a usable provider, and that a
// pass-through HTTP middleware leaves the wrapped handler intact.
package telemetry
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"github.com/owncord/server/config"
)
func TestInitNoOpReturnsShutdown(t *testing.T) {
shutdown, err := Init(context.Background(), config.TelemetryConfig{Enabled: false})
if err != nil {
t.Fatalf("Init: %v", err)
}
if shutdown == nil {
t.Fatal("expected non-nil shutdown")
}
if err := shutdown(context.Background()); err != nil {
t.Fatalf("shutdown: %v", err)
}
}
func TestGlobalReturnsUsableProvider(t *testing.T) {
p := Global()
if p == nil {
t.Fatal("Global returned nil")
}
tracer := p.Tracer("test")
_, span := tracer.Start(context.Background(), "noop")
span.SetAttributes(String("k", "v"))
span.End()
meter := p.Meter("test")
meter.Counter("c", "").Add(context.Background(), 1)
meter.Histogram("h", "", "s").Record(context.Background(), 1.0)
meter.Gauge("g", "").Set(context.Background(), 0.5)
}
func TestHTTPMiddlewareIsPassThrough(t *testing.T) {
called := false
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
called = true
w.WriteHeader(http.StatusTeapot)
})
wrapped := HTTPMiddleware()(handler)
rec := httptest.NewRecorder()
wrapped.ServeHTTP(rec, httptest.NewRequest("GET", "/", nil))
if !called {
t.Fatal("inner handler not called")
}
if rec.Code != http.StatusTeapot {
t.Fatalf("status: got %d want %d", rec.Code, http.StatusTeapot)
}
}
func TestPrometheusHandlerNilByDefault(t *testing.T) {
if PrometheusHandler() != nil {
t.Fatal("expected nil Prometheus handler in no-op build")
}
}
+157
View File
@@ -0,0 +1,157 @@
// Phase B Step 7 — Event Persistence Layer.
//
// EventPersister is an asynchronous batched writer that drains broadcast
// events from an in-memory channel into the EventStore. It must never block
// the broadcast hot path: when the queue is full, events are dropped and a
// counter is incremented. The reconnection handler tolerates gaps because the
// in-memory ring buffer remains the primary cold-start source for clients
// whose last_seq is recent.
package ws
import (
"context"
"log/slog"
"sync"
"sync/atomic"
"time"
"github.com/owncord/server/store"
)
// pendingEvent is a single event waiting to be flushed to the EventStore.
type pendingEvent struct {
eventType string
channelID int64
payload []byte
}
// EventPersister batches broadcast events and writes them to an EventStore.
type EventPersister struct {
store store.EventStore
queue chan pendingEvent
batchSize int
flushEvy time.Duration
stopOnce sync.Once
stop chan struct{}
done chan struct{}
persisted atomic.Uint64
dropped atomic.Uint64
flushes atomic.Uint64
errors atomic.Uint64
}
// NewEventPersister returns a persister wired to s. queueSize sets the
// channel buffer; once full, Enqueue increments the dropped counter without
// blocking. batchSize and flushEvery control the flush triggers.
func NewEventPersister(s store.EventStore, queueSize, batchSize int, flushEvery time.Duration) *EventPersister {
if queueSize <= 0 {
queueSize = 1024
}
if batchSize <= 0 {
batchSize = 50
}
if flushEvery <= 0 {
flushEvery = 100 * time.Millisecond
}
return &EventPersister{
store: s,
queue: make(chan pendingEvent, queueSize),
batchSize: batchSize,
flushEvy: flushEvery,
stop: make(chan struct{}),
done: make(chan struct{}),
}
}
// Start launches the background flusher goroutine.
func (p *EventPersister) Start(ctx context.Context) {
go p.run(ctx)
}
// Enqueue queues an event for persistence. Non-blocking; drops on full queue.
func (p *EventPersister) Enqueue(eventType string, channelID int64, payload []byte) {
if p == nil {
return
}
// Defensive copy: callers may reuse buffers.
cp := make([]byte, len(payload))
copy(cp, payload)
select {
case p.queue <- pendingEvent{eventType: eventType, channelID: channelID, payload: cp}:
default:
p.dropped.Add(1)
}
}
// Stop signals the persister to drain remaining events and exit. Blocks until
// the goroutine exits or ctx is cancelled.
func (p *EventPersister) Stop(ctx context.Context) {
if p == nil {
return
}
p.stopOnce.Do(func() { close(p.stop) })
select {
case <-p.done:
case <-ctx.Done():
}
}
// Stats returns lifetime counters.
func (p *EventPersister) Stats() (persisted, dropped, flushes, errs uint64) {
return p.persisted.Load(), p.dropped.Load(), p.flushes.Load(), p.errors.Load()
}
func (p *EventPersister) run(ctx context.Context) {
defer close(p.done)
tick := time.NewTicker(p.flushEvy)
defer tick.Stop()
batch := make([]pendingEvent, 0, p.batchSize)
flush := func() {
if len(batch) == 0 {
return
}
p.flushes.Add(1)
for _, evt := range batch {
if _, err := p.store.PersistEvent(ctx, evt.eventType, evt.channelID, evt.payload); err != nil {
p.errors.Add(1)
slog.Warn("event persister: PersistEvent failed",
"event_type", evt.eventType, "channel_id", evt.channelID, "err", err)
continue
}
p.persisted.Add(1)
}
batch = batch[:0]
}
for {
select {
case <-p.stop:
// Drain anything still in the channel before exiting.
for {
select {
case evt := <-p.queue:
batch = append(batch, evt)
if len(batch) >= p.batchSize {
flush()
}
default:
flush()
return
}
}
case <-ctx.Done():
flush()
return
case evt := <-p.queue:
batch = append(batch, evt)
if len(batch) >= p.batchSize {
flush()
}
case <-tick.C:
flush()
}
}
}
+90
View File
@@ -0,0 +1,90 @@
// Phase B Step 7 — EventPersister behavioural test.
//
// Confirms that:
// - events flow into the configured store via the batched flusher,
// - dropped events are counted on full queue,
// - Stop() drains pending events before exiting.
package ws
import (
"context"
"testing"
"time"
"github.com/owncord/server/store"
)
func TestEventPersisterFlushesBatch(t *testing.T) {
mem := store.NewMemStore()
p := NewEventPersister(mem, 1024, 4, 50*time.Millisecond)
ctx := context.Background()
p.Start(ctx)
t.Cleanup(func() { p.Stop(ctx) })
for i := 0; i < 10; i++ {
p.Enqueue("broadcast", 0, []byte(`{"type":"x"}`))
}
// Wait for at least one flush tick.
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
persisted, _, _, _ := p.Stats()
if persisted >= 10 {
break
}
time.Sleep(10 * time.Millisecond)
}
persisted, dropped, _, errs := p.Stats()
if persisted != 10 {
t.Fatalf("expected 10 persisted, got %d", persisted)
}
if dropped != 0 {
t.Fatalf("expected 0 dropped, got %d", dropped)
}
if errs != 0 {
t.Fatalf("expected 0 errors, got %d", errs)
}
// Verify the events landed in the store.
rows, err := mem.GetEventsSince(ctx, 0, 100)
if err != nil {
t.Fatal(err)
}
if len(rows) != 10 {
t.Fatalf("expected 10 rows in mem store, got %d", len(rows))
}
}
func TestEventPersisterDropsOnFullQueue(t *testing.T) {
mem := store.NewMemStore()
// Tiny queue, very long flush interval — guarantees drops because the
// flusher won't drain fast enough.
p := NewEventPersister(mem, 2, 1024, time.Hour)
// NB: Start is intentionally NOT called so the queue stays full.
for i := 0; i < 50; i++ {
p.Enqueue("broadcast", 0, []byte(`{}`))
}
_, dropped, _, _ := p.Stats()
if dropped == 0 {
t.Fatal("expected drops with full queue and no consumer")
}
}
func TestEventPersisterStopDrains(t *testing.T) {
mem := store.NewMemStore()
p := NewEventPersister(mem, 256, 100, time.Hour)
p.Start(context.Background())
for i := 0; i < 5; i++ {
p.Enqueue("broadcast", 0, []byte(`{}`))
}
stopCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
p.Stop(stopCtx)
persisted, _, _, _ := p.Stats()
if persisted != 5 {
t.Fatalf("expected 5 persisted after Stop drain, got %d", persisted)
}
}
+62
View File
@@ -0,0 +1,62 @@
// Phase B Step 7 — Event Persistence Layer (pruner).
//
// StartEventPruner runs a background goroutine that deletes events older than
// the configured retention window. It is the bounded-storage half of the
// event persistence design: the persister appends, the pruner trims.
package ws
import (
"context"
"log/slog"
"time"
"github.com/owncord/server/store"
)
// StartEventPruner launches a goroutine that wakes every interval and deletes
// events older than retention. The goroutine exits when ctx is cancelled.
func StartEventPruner(ctx context.Context, s store.EventStore, retention, interval time.Duration) {
if s == nil {
return
}
if retention <= 0 {
retention = 24 * time.Hour
}
if interval <= 0 {
interval = time.Hour
}
go func() {
// Run once shortly after startup so a tiny dataset stays small.
startupDelay := time.NewTimer(time.Minute)
defer startupDelay.Stop()
select {
case <-ctx.Done():
return
case <-startupDelay.C:
}
runPrune(ctx, s, retention)
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
runPrune(ctx, s, retention)
}
}
}()
}
func runPrune(ctx context.Context, s store.EventStore, retention time.Duration) {
cutoff := time.Now().Add(-retention)
deleted, err := s.PruneEventsOlderThan(ctx, cutoff)
if err != nil {
slog.Warn("event pruner: PruneEventsOlderThan failed", "err", err)
return
}
if deleted > 0 {
slog.Info("event pruner: pruned old events", "deleted", deleted, "cutoff", cutoff.Format(time.RFC3339))
}
}
+45
View File
@@ -14,6 +14,7 @@ import (
"github.com/owncord/server/db"
"github.com/owncord/server/permissions"
"github.com/owncord/server/service"
"github.com/owncord/server/store"
"github.com/owncord/server/syncutil"
)
@@ -49,6 +50,15 @@ type Hub struct {
replayBuf *EventRingBuffer // recent broadcast events for reconnection replay
broadcastDrops atomic.Uint64 // counts messages dropped due to full broadcast channel
// Phase B Step 7 — event persistence. nil = ring buffer only.
eventPersister *EventPersister
eventStore store.EventStore // read path for cold-tier replay
// Phase B Step 7 — reconnection tier metrics. Incremented per resume.
reconnectTierBuf atomic.Uint64
reconnectTierDB atomic.Uint64
reconnectTierFull atomic.Uint64
// Settings cache — avoids per-connection DB queries for server_name/motd.
settingsMu syncutil.RWMutex
settingsName string
@@ -538,6 +548,7 @@ func (h *Hub) sendSequencedToUsers(channelID int64, userIDs []int64, msg []byte)
// Store DM event for reconnect replay; filtering is channel-based and uses
// allowed channel IDs computed at auth time (including open DMs).
h.replayBuf.Push(seq, channelID, wrapped)
h.persistEvent(channelID, wrapped)
for _, userID := range userIDs {
h.SendToUser(userID, wrapped)
@@ -552,6 +563,7 @@ func (h *Hub) sendSequencedToUsersHigh(channelID int64, userIDs []int64, msg []b
seq := h.nextSeq()
wrapped := wrapWithSeq(msg, seq)
h.replayBuf.Push(seq, channelID, wrapped)
h.persistEvent(channelID, wrapped)
for _, userID := range userIDs {
h.SendToUserHigh(userID, wrapped)
@@ -607,6 +619,38 @@ func (h *Hub) ReplayBuffer() *EventRingBuffer {
return h.replayBuf
}
// SetEventPersister attaches a persister so subsequent broadcasts are also
// written to the persistent EventStore. Pass nil to disable.
func (h *Hub) SetEventPersister(p *EventPersister) {
h.eventPersister = p
}
// SetEventStore attaches a read-side EventStore used by the cold-tier
// reconnect replay path. Typically the same store backing SetEventPersister.
func (h *Hub) SetEventStore(s store.EventStore) {
h.eventStore = s
}
// ReconnectTierStats returns the per-tier resume hit counters in the order
// (buffer, db, full). Phase B Step 7 metrics surface; OpenTelemetry meters
// (Step 8) read from the same atomics.
func (h *Hub) ReconnectTierStats() (buffer, db, full uint64) {
return h.reconnectTierBuf.Load(), h.reconnectTierDB.Load(), h.reconnectTierFull.Load()
}
// persistEvent enqueues a broadcast event for cold-storage persistence. Safe
// to call with a nil persister; never blocks the broadcast hot path.
func (h *Hub) persistEvent(channelID int64, payload []byte) {
if h.eventPersister == nil {
return
}
eventType := "broadcast"
if channelID != 0 {
eventType = "channel_broadcast"
}
h.eventPersister.Enqueue(eventType, channelID, payload)
}
// wrapWithSeq injects a "seq" field into a JSON message without re-serializing.
func wrapWithSeq(msg []byte, seq uint64) []byte {
// Fast path: inject seq after the opening brace.
@@ -756,6 +800,7 @@ func (h *Hub) deliverBroadcast(bm broadcastMsg) {
// Store in replay buffer for reconnection recovery.
h.replayBuf.Push(seq, bm.channelID, msg)
h.persistEvent(bm.channelID, msg)
if bm.channelID == 0 {
// Global broadcast — deliver to every connected client.
+35 -2
View File
@@ -15,6 +15,7 @@ import (
"github.com/owncord/server/config"
"github.com/owncord/server/db"
"github.com/owncord/server/permissions"
"github.com/owncord/server/telemetry"
)
const (
@@ -121,9 +122,41 @@ func (h *Hub) handleReconnect(
}
events := h.ReplayBuffer().EventsSinceFiltered(lastSeq, allowedChannelIDs)
replaySource := "buffer"
if events == nil {
return false
// Phase B Step 7 — try cold-tier replay from the EventStore before
// giving up and forcing a full ready re-sync.
if h.eventStore != nil {
channelIDs := make([]int64, 0, len(allowedChannelIDs))
for cid := range allowedChannelIDs {
channelIDs = append(channelIDs, cid)
}
const maxColdReplay = 5000
persisted, dbErr := h.eventStore.GetEventsSinceForChannels(ctx, int64(lastSeq), channelIDs, maxColdReplay)
if dbErr != nil {
slog.Warn("ws handleReconnect: cold-tier replay query failed",
"user_id", c.userID, "err", dbErr)
} else if len(persisted) > 0 {
events = make([][]byte, 0, len(persisted))
for _, p := range persisted {
events = append(events, p.Payload)
}
replaySource = "db"
}
}
if events == nil {
h.reconnectTierFull.Add(1)
telemetry.NewAppMetrics().WSReconnectTierTotal.Add(ctx, 1, telemetry.String("tier", "full"))
return false
}
}
switch replaySource {
case "buffer":
h.reconnectTierBuf.Add(1)
case "db":
h.reconnectTierDB.Add(1)
}
telemetry.NewAppMetrics().WSReconnectTierTotal.Add(ctx, 1, telemetry.String("tier", replaySource))
// Register BEFORE writing replay data so broadcasts that arrive during
// the write window are queued in the client's send buffer instead of
@@ -147,7 +180,7 @@ func (h *Hub) handleReconnect(
return true
}
}
slog.Info("ws replay completed", "user_id", c.userID, "events_replayed", len(events), "from_seq", lastSeq)
slog.Info("ws replay completed", "user_id", c.userID, "events_replayed", len(events), "from_seq", lastSeq, "source", replaySource)
// Update presence but skip member_join — user was already known.
if updateErr := database.UpdateUserStatus(c.userID, "online"); updateErr != nil {