diff --git a/Client/tauri-client/package.json b/Client/tauri-client/package.json index 03d28861..7b8e2f1c 100644 --- a/Client/tauri-client/package.json +++ b/Client/tauri-client/package.json @@ -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" } } diff --git a/Client/tauri-client/src/components/solid/Badge.test.tsx b/Client/tauri-client/src/components/solid/Badge.test.tsx new file mode 100644 index 00000000..cb4e5397 --- /dev/null +++ b/Client/tauri-client/src/components/solid/Badge.test.tsx @@ -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(() => ); + expect(getByText("online")).toBeTruthy(); + }); + + it("invokes onClick when activated", () => { + let clicked = 0; + const { getByText } = render(() => ( + clicked++} /> + )); + getByText("press me").click(); + expect(clicked).toBe(1); + }); +}); diff --git a/Client/tauri-client/src/components/solid/Badge.tsx b/Client/tauri-client/src/components/solid/Badge.tsx new file mode 100644 index 00000000..32570aef --- /dev/null +++ b/Client/tauri-client/src/components/solid/Badge.tsx @@ -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, 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 ( + { + if (props.onClick && (e.key === "Enter" || e.key === " ")) { + e.preventDefault(); + props.onClick(); + } + }} + > + {props.label} + + ); +} diff --git a/Client/tauri-client/src/components/solid/ChannelListItem.tsx b/Client/tauri-client/src/components/solid/ChannelListItem.tsx new file mode 100644 index 00000000..60bf4b4c --- /dev/null +++ b/Client/tauri-client/src/components/solid/ChannelListItem.tsx @@ -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( + channelsStore, + (s) => s.channels.get(props.channelId), + ); + const isActive = fromStoreSlice( + channelsStore, + (s) => s.activeChannelId === props.channelId, + ); + + return ( + + {(ch) => ( +
  • props.onSelect(props.channelId)} + > + # + {ch().name} + 0}> + + +
    • + )} +     + ); +} diff --git a/Client/tauri-client/src/components/solid/PluginContainer.tsx b/Client/tauri-client/src/components/solid/PluginContainer.tsx new file mode 100644 index 00000000..d032bd57 --- /dev/null +++ b/Client/tauri-client/src/components/solid/PluginContainer.tsx @@ -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 ( +
    + ); +} diff --git a/Client/tauri-client/src/components/solid/README.md b/Client/tauri-client/src/components/solid/README.md new file mode 100644 index 00000000..ed0b86af --- /dev/null +++ b/Client/tauri-client/src/components/solid/README.md @@ -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. diff --git a/Client/tauri-client/src/lib/pluginBridge.ts b/Client/tauri-client/src/lib/pluginBridge.ts new file mode 100644 index 00000000..16b17e02 --- /dev/null +++ b/Client/tauri-client/src/lib/pluginBridge.ts @@ -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(); + private listeners = new Set(); + private themeVars: Record = {}; + + constructor() { + window.addEventListener("message", this.onMessage); + } + + /** Replace the theme variables broadcast to plugin iframes. */ + setTheme(vars: Record): 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(); diff --git a/Client/tauri-client/src/lib/solidAdapter.ts b/Client/tauri-client/src/lib/solidAdapter.ts new file mode 100644 index 00000000..5c35df7c --- /dev/null +++ b/Client/tauri-client/src/lib/solidAdapter.ts @@ -0,0 +1,56 @@ +/** + * Phase B Step 6 — Solid.js adapter. + * + * Bridges the existing custom reactive `Store` (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
    {auth().username}
    ; + * + * 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(store: Store): Accessor { + const [value, setValue] = createSignal(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( + store: Store, + selector: (state: T) => S, + isEqual?: (a: S, b: S) => boolean, +): Accessor { + const initial = selector(store.getState()); + const [value, setValue] = createSignal(initial, { equals: false }); + const unsub = store.subscribeSelector( + selector, + (next) => setValue(() => next), + isEqual, + ); + onCleanup(unsub); + return value; +} diff --git a/Client/tauri-client/src/lib/solidMount.ts b/Client/tauri-client/src/lib/solidMount.ts new file mode 100644 index 00000000..24c32590 --- /dev/null +++ b/Client/tauri-client/src/lib/solidMount.ts @@ -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(); + }, + }; +} diff --git a/Client/tauri-client/tsconfig.json b/Client/tauri-client/tsconfig.json index 784de97f..9ea85696 100644 --- a/Client/tauri-client/tsconfig.json +++ b/Client/tauri-client/tsconfig.json @@ -11,6 +11,8 @@ "resolveJsonModule": true, "isolatedModules": true, "noEmit": true, + "jsx": "preserve", + "jsxImportSource": "solid-js", "lib": [ "ES2023", "DOM", diff --git a/Client/tauri-client/vite.config.ts b/Client/tauri-client/vite.config.ts index 6f58351a..8f5a5449 100644 --- a/Client/tauri-client/vite.config.ts +++ b/Client/tauri-client/vite.config.ts @@ -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, diff --git a/PHASE_BC_LOCAL_TODO.md b/PHASE_BC_LOCAL_TODO.md new file mode 100644 index 00000000..21093168 --- /dev/null +++ b/PHASE_BC_LOCAL_TODO.md @@ -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). diff --git a/Server/api/plugins_handler.go b/Server/api/plugins_handler.go new file mode 100644 index 00000000..1ed8351d --- /dev/null +++ b/Server/api/plugins_handler.go @@ -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 +} diff --git a/Server/api/router.go b/Server/api/router.go index b4f5174d..a80bd59e 100644 --- a/Server/api/router.go +++ b/Server/api/router.go @@ -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). diff --git a/Server/config/config.go b/Server/config/config.go index e0f7835d..f1a194b0 100644 --- a/Server/config/config.go +++ b/Server/config/config.go @@ -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{}, + }, } } diff --git a/Server/db/persisted_event.go b/Server/db/persisted_event.go new file mode 100644 index 00000000..85cf9eb5 --- /dev/null +++ b/Server/db/persisted_event.go @@ -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 +} diff --git a/Server/db/queries/postgres/events.sql b/Server/db/queries/postgres/events.sql new file mode 100644 index 00000000..2c61d7fa --- /dev/null +++ b/Server/db/queries/postgres/events.sql @@ -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; diff --git a/Server/db/queries/postgres/plugins.sql b/Server/db/queries/postgres/plugins.sql new file mode 100644 index 00000000..d0349d86 --- /dev/null +++ b/Server/db/queries/postgres/plugins.sql @@ -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; diff --git a/Server/db/queries/sqlite/events.sql b/Server/db/queries/sqlite/events.sql new file mode 100644 index 00000000..e6a18941 --- /dev/null +++ b/Server/db/queries/sqlite/events.sql @@ -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 < ?; diff --git a/Server/db/queries/sqlite/plugins.sql b/Server/db/queries/sqlite/plugins.sql new file mode 100644 index 00000000..b3825d69 --- /dev/null +++ b/Server/db/queries/sqlite/plugins.sql @@ -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 = ?; diff --git a/Server/main.go b/Server/main.go index 75da7008..a87c3e6b 100644 --- a/Server/main.go +++ b/Server/main.go @@ -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{ diff --git a/Server/migrations/014_events_table.sql b/Server/migrations/014_events_table.sql new file mode 100644 index 00000000..39b13fe9 --- /dev/null +++ b/Server/migrations/014_events_table.sql @@ -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); diff --git a/Server/migrations/015_plugins.sql b/Server/migrations/015_plugins.sql new file mode 100644 index 00000000..7552aaba --- /dev/null +++ b/Server/migrations/015_plugins.sql @@ -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) +); diff --git a/Server/migrations/postgres/001_initial_schema.sql b/Server/migrations/postgres/001_initial_schema.sql index b4ec48bc..07fc8ab5 100644 --- a/Server/migrations/postgres/001_initial_schema.sql +++ b/Server/migrations/postgres/001_initial_schema.sql @@ -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) +); diff --git a/Server/plugin/errors.go b/Server/plugin/errors.go new file mode 100644 index 00000000..fbec2ba3 --- /dev/null +++ b/Server/plugin/errors.go @@ -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") diff --git a/Server/plugin/examples/hello/README.md b/Server/plugin/examples/hello/README.md new file mode 100644 index 00000000..921c6df3 --- /dev/null +++ b/Server/plugin/examples/hello/README.md @@ -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. diff --git a/Server/plugin/examples/hello/plugin.json b/Server/plugin/examples/hello/plugin.json new file mode 100644 index 00000000..7ed04242 --- /dev/null +++ b/Server/plugin/examples/hello/plugin.json @@ -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 + } +} diff --git a/Server/plugin/host_commands.go b/Server/plugin/host_commands.go new file mode 100644 index 00000000..9bd104da --- /dev/null +++ b/Server/plugin/host_commands.go @@ -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) +} diff --git a/Server/plugin/host_events.go b/Server/plugin/host_events.go new file mode 100644 index 00000000..82af0c18 --- /dev/null +++ b/Server/plugin/host_events.go @@ -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 + } +} diff --git a/Server/plugin/host_http.go b/Server/plugin/host_http.go new file mode 100644 index 00000000..39fbd0ee --- /dev/null +++ b/Server/plugin/host_http.go @@ -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 +} diff --git a/Server/plugin/host_storage.go b/Server/plugin/host_storage.go new file mode 100644 index 00000000..812bbed9 --- /dev/null +++ b/Server/plugin/host_storage.go @@ -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) +} diff --git a/Server/plugin/host_ui.go b/Server/plugin/host_ui.go new file mode 100644 index 00000000..73929853 --- /dev/null +++ b/Server/plugin/host_ui.go @@ -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//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) + }) +} diff --git a/Server/plugin/loader.go b/Server/plugin/loader.go new file mode 100644 index 00000000..1d6bc1d3 --- /dev/null +++ b/Server/plugin/loader.go @@ -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 +} diff --git a/Server/plugin/manifest.go b/Server/plugin/manifest.go new file mode 100644 index 00000000..2f6be965 --- /dev/null +++ b/Server/plugin/manifest.go @@ -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 +} diff --git a/Server/plugin/plugin_test.go b/Server/plugin/plugin_test.go new file mode 100644 index 00000000..8f9cc216 --- /dev/null +++ b/Server/plugin/plugin_test.go @@ -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) + } +} diff --git a/Server/plugin/registry.go b/Server/plugin/registry.go new file mode 100644 index 00000000..df8cdaab --- /dev/null +++ b/Server/plugin/registry.go @@ -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 +} diff --git a/Server/plugin/sandbox_default.go b/Server/plugin/sandbox_default.go new file mode 100644 index 00000000..1ae343d9 --- /dev/null +++ b/Server/plugin/sandbox_default.go @@ -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 +} diff --git a/Server/plugin/sandbox_wazero.go b/Server/plugin/sandbox_wazero.go new file mode 100644 index 00000000..0811e483 --- /dev/null +++ b/Server/plugin/sandbox_wazero.go @@ -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 +} diff --git a/Server/service/channel.go b/Server/service/channel.go index c6e84f93..f77c2d49 100644 --- a/Server/service/channel.go +++ b/Server/service/channel.go @@ -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) diff --git a/Server/service/message.go b/Server/service/message.go index 0a78ee15..d6bc6df3 100644 --- a/Server/service/message.go +++ b/Server/service/message.go @@ -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) { diff --git a/Server/service/permission.go b/Server/service/permission.go index 018d3c22..2671ba7d 100644 --- a/Server/service/permission.go +++ b/Server/service/permission.go @@ -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 diff --git a/Server/store/memstore.go b/Server/store/memstore.go index 109c0940..549ed96e 100644 --- a/Server/store/memstore.go +++ b/Server/store/memstore.go @@ -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. diff --git a/Server/store/memstore_events.go b/Server/store/memstore_events.go new file mode 100644 index 00000000..3213cd5d --- /dev/null +++ b/Server/store/memstore_events.go @@ -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 +} diff --git a/Server/store/postgres.go b/Server/store/postgres.go index 74ff78a4..a1dead5f 100644 --- a/Server/store/postgres.go +++ b/Server/store/postgres.go @@ -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) diff --git a/Server/store/sqlite_events.go b/Server/store/sqlite_events.go new file mode 100644 index 00000000..6b54ff97 --- /dev/null +++ b/Server/store/sqlite_events.go @@ -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{} +} diff --git a/Server/store/store.go b/Server/store/store.go index 0e77d0c9..1f968bdc 100644 --- a/Server/store/store.go +++ b/Server/store/store.go @@ -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) +} diff --git a/Server/telemetry/metrics.go b/Server/telemetry/metrics.go new file mode 100644 index 00000000..47f3968f --- /dev/null +++ b/Server/telemetry/metrics.go @@ -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 +} diff --git a/Server/telemetry/middleware.go b/Server/telemetry/middleware.go new file mode 100644 index 00000000..e20dca7d --- /dev/null +++ b/Server/telemetry/middleware.go @@ -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() } diff --git a/Server/telemetry/telemetry.go b/Server/telemetry/telemetry.go new file mode 100644 index 00000000..d7c84a69 --- /dev/null +++ b/Server/telemetry/telemetry.go @@ -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{}) +} diff --git a/Server/telemetry/telemetry_default.go b/Server/telemetry/telemetry_default.go new file mode 100644 index 00000000..6e632aa3 --- /dev/null +++ b/Server/telemetry/telemetry_default.go @@ -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 +} diff --git a/Server/telemetry/telemetry_otel.go b/Server/telemetry/telemetry_otel.go new file mode 100644 index 00000000..87b3a7ea --- /dev/null +++ b/Server/telemetry/telemetry_otel.go @@ -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 } diff --git a/Server/telemetry/telemetry_test.go b/Server/telemetry/telemetry_test.go new file mode 100644 index 00000000..97625bfd --- /dev/null +++ b/Server/telemetry/telemetry_test.go @@ -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") + } +} diff --git a/Server/ws/event_persister.go b/Server/ws/event_persister.go new file mode 100644 index 00000000..726896e4 --- /dev/null +++ b/Server/ws/event_persister.go @@ -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() + } + } +} diff --git a/Server/ws/event_persister_test.go b/Server/ws/event_persister_test.go new file mode 100644 index 00000000..2e54dc07 --- /dev/null +++ b/Server/ws/event_persister_test.go @@ -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) + } +} diff --git a/Server/ws/event_pruner.go b/Server/ws/event_pruner.go new file mode 100644 index 00000000..73d032b4 --- /dev/null +++ b/Server/ws/event_pruner.go @@ -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)) + } +} diff --git a/Server/ws/hub.go b/Server/ws/hub.go index 1bfc65a4..0856a2b5 100644 --- a/Server/ws/hub.go +++ b/Server/ws/hub.go @@ -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. diff --git a/Server/ws/serve.go b/Server/ws/serve.go index 6f022542..546baa6f 100644 --- a/Server/ws/serve.go +++ b/Server/ws/serve.go @@ -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 {