diff --git a/apps/backend/cmd/plugin-fixture/fixture-package/manifest.yaml b/apps/backend/cmd/plugin-fixture/fixture-package/manifest.yaml index 86e7799bd7..03c96d062a 100644 --- a/apps/backend/cmd/plugin-fixture/fixture-package/manifest.yaml +++ b/apps/backend/cmd/plugin-fixture/fixture-package/manifest.yaml @@ -26,6 +26,7 @@ capabilities: events: ["task.created"] state: true secrets: true + user_state: true webhooks: - key: "test-hook" diff --git a/apps/backend/cmd/plugin-fixture/fixture-package/ui/bundle.js b/apps/backend/cmd/plugin-fixture/fixture-package/ui/bundle.js index 952c4d75f6..643342bcb2 100644 --- a/apps/backend/cmd/plugin-fixture/fixture-package/ui/bundle.js +++ b/apps/backend/cmd/plugin-fixture/fixture-package/ui/bundle.js @@ -9,6 +9,13 @@ * default `mod+shift+j`) which opens a `host.openModal(...)` demo modal. * Uses only host.React/host.jsx. * + * It also registers the plugin-hooks surface this fixture exists to drive + * end to end: a "Notes" task panel (registerTaskPanel, mobile-enabled) that + * round-trips a single per-user document through host.storage + * (get on mount, debounced set, subscribe to pick up a write from another + * tab/surface), a task-card-indicators slot component, and a + * registerTaskMenuAction under the kanban card's "edit" group. + * * The task-created counter lives in module scope (not component state) with * a tiny listener set, so it survives across route navigations (the page * component unmounts/remounts as the user navigates away and back). @@ -66,6 +73,106 @@ return jsx("span", { id: "hello-main-top-bar" }, "Hello " + slotProps.currentPage); } + // Debounce delay for the Notes panel's autosave — short, so e2e specs + // don't need to wait long for a write to reach host.storage. + var NOTES_SAVE_DEBOUNCE_MS = 150; + + function useNotesValue(taskId, panelId) { + var state = React.useState(""); + var value = state[0]; + var setValue = state[1]; + var loadedState = React.useState(false); + var loaded = loadedState[0]; + var setLoaded = loadedState[1]; + + React.useEffect( + function () { + var cancelled = false; + // Each notification (subscribe's refresh callback, plus the initial + // call) starts its own host.storage.get. Two in flight at once can + // resolve out of order, so `cancelled` alone (unmount-only) isn't + // enough — an older read landing after a newer one would overwrite + // it. Only the read matching the current generation may commit. + var refreshGeneration = 0; + function refresh() { + var generation = ++refreshGeneration; + host.storage.get("task", taskId, "note").then( + function (entry) { + if (cancelled || generation !== refreshGeneration) return; + setValue(entry ? entry.value : ""); + setLoaded(true); + }, + function () { + if (cancelled || generation !== refreshGeneration) return; + setLoaded(true); + }, + ); + } + refresh(); + // Scope echo suppression to this panel instance (panelId) rather + // than the host's shared tab-wide default writerId. The kanban + // "Enhance notes" action below writes this same note using that + // shared default (it has no ongoing subscription of its own to + // protect) — if this panel used the same default, the action's + // write would look like this panel's own echo and get silently + // dropped instead of refreshing the panel. + var unsubscribe = host.storage.subscribe( + { scope: "task", scopeId: taskId, key: "note", writerId: panelId }, + refresh, + ); + return function () { + cancelled = true; + unsubscribe(); + }; + }, + [taskId, panelId], + ); + + return [value, setValue, loaded]; + } + + function NotesPanel(props) { + var taskId = props.taskId; + var panelId = props.panelId; + var notesValue = useNotesValue(taskId, panelId); + var value = notesValue[0]; + var setValue = notesValue[1]; + var loaded = notesValue[2]; + var debounceRef = React.useRef(null); + + function handleChange(e) { + var next = e.target.value; + setValue(next); + if (debounceRef.current) clearTimeout(debounceRef.current); + debounceRef.current = setTimeout(function () { + host.storage + .set("task", taskId, "note", next, { writerId: panelId }) + .catch(function () { + // surface the failed autosave instead of dropping it silently + }); + }, NOTES_SAVE_DEBOUNCE_MS); + } + + if (!loaded) { + return jsx("div", { "data-testid": "e2e-notes-panel-loading" }, "Loading notes…"); + } + return jsx("textarea", { + "data-testid": "e2e-notes-panel", + "data-presentation": props.presentation, + value: value, + onChange: handleChange, + }); + } + + function CardIndicator(props) { + var slotProps = props.slotProps || {}; + return jsx( + "span", + { "data-testid": "e2e-card-indicator", "data-task-id": slotProps.taskId }, + "N", + ); + } + function StatusSlot(props) { var slotProps = props.slotProps || {}; var id = slotProps.placement === "left" ? "hello-status-left" : "hello-status-right"; @@ -94,6 +201,34 @@ incrementCount(); }); + registry.registerTaskPanel({ + id: "notes", + title: "Notes", + icon: "book", + Component: NotesPanel, + mobileEnabled: true, + }); + registry.registerComponent("task-card-indicators", CardIndicator); + registry.registerTaskMenuAction({ + id: "enhance-notes", + label: "Enhance notes", + group: "edit", + run: function (context) { + return host.storage + .set("task", context.taskId, "note", "Enhanced via plugin action") + .then(function () { + // Keep the received responsive context observable to the E2E + // fixture without changing the plugin-facing action contract. + return host.storage.set( + "task", + context.taskId, + "menu-presentation", + context.presentation, + ); + }); + }, + }); + registry.registerKeybinding("open-demo", function () { function DemoModalContent() { return jsx( diff --git a/apps/backend/cmd/plugin-fixture/fixture_package_test.go b/apps/backend/cmd/plugin-fixture/fixture_package_test.go index 5e7c501745..261e710571 100644 --- a/apps/backend/cmd/plugin-fixture/fixture_package_test.go +++ b/apps/backend/cmd/plugin-fixture/fixture_package_test.go @@ -28,6 +28,7 @@ func TestFixtureManifest_ParsesAndValidates(t *testing.T) { require.Equal(t, "/ui/bundle.js", m.UI.Bundle) require.True(t, m.HasEvent("task.created")) require.True(t, m.Capabilities.State) + require.True(t, m.Capabilities.UserState) require.Len(t, m.Webhooks, 1) require.Equal(t, "test-hook", m.Webhooks[0].Key) diff --git a/apps/backend/internal/auth/httpmw/middleware_test.go b/apps/backend/internal/auth/httpmw/middleware_test.go index 3e1815de51..bd60bdf83d 100644 --- a/apps/backend/internal/auth/httpmw/middleware_test.go +++ b/apps/backend/internal/auth/httpmw/middleware_test.go @@ -186,6 +186,10 @@ func TestEnabledModeAllowlistMatrix(t *testing.T) { {name: "workspaces api", method: http.MethodGet, path: "/api/v1/workspaces", blocked: true}, {name: "plugin management", method: http.MethodGet, path: "/api/plugins", blocked: true}, {name: "plugin bundle unauthenticated", method: http.MethodGet, path: "/api/plugins/p1/bundle", blocked: true}, + { + name: "plugin user-state unauthenticated", method: http.MethodGet, + path: "/api/plugins/p1/user-state/task/task1/note", blocked: true, + }, {name: "debug", method: http.MethodGet, path: "/debug/vars", blocked: true}, } for _, tc := range cases { diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 1e20f44025..6ce6ec775a 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -158,6 +158,14 @@ const ( UserSettingsUpdated = "user.settings.updated" ) +// PluginUserStateUpdated fires after a successful write/delete on a +// plugin's per-user storage route (/api/plugins/:id/user-state/...). The +// event payload carries only keys (pluginId/scope/scopeId/key/writerId), +// never the stored value — see +// docs/decisions/2026-08-01-per-user-plugin-storage.md. Routed to the +// writing user's own WS connections only, via UserEventBroadcaster. +const PluginUserStateUpdated = "plugin.user-state.updated" + // Event types for system maintenance jobs (VACUUM, factory reset, snapshot // create/restore, disk walk). Published by internal/system/jobs.Tracker on // every state transition and broadcast to all WebSocket clients. diff --git a/apps/backend/internal/gateway/websocket/user_notifications.go b/apps/backend/internal/gateway/websocket/user_notifications.go index 3a978c512f..47a179d8e1 100644 --- a/apps/backend/internal/gateway/websocket/user_notifications.go +++ b/apps/backend/internal/gateway/websocket/user_notifications.go @@ -2,6 +2,7 @@ package websocket import ( "context" + "encoding/json" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" @@ -26,6 +27,7 @@ func RegisterUserNotifications(ctx context.Context, eventBus bus.EventBus, hub * } b.subscribe(eventBus, events.UserSettingsUpdated, ws.ActionUserSettingsUpdated) + b.subscribe(eventBus, events.PluginUserStateUpdated, ws.ActionPluginUserStateUpdated) go func() { <-ctx.Done() @@ -46,17 +48,11 @@ func (b *UserEventBroadcaster) Close() { func (b *UserEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action string) { sub, err := eventBus.Subscribe(subject, func(ctx context.Context, event *bus.Event) error { - // Try to extract user_id from event data (works for both map and struct types) - var userID string - if data, ok := event.Data.(map[string]interface{}); ok { - userID, _ = data["user_id"].(string) - } else if data, ok := event.Data.(interface{ GetUserID() string }); ok { - userID = data.GetUserID() - } + userID, payload := routingUserIDAndPayload(event.Data) if userID == "" { return nil } - msg, err := ws.NewNotification(action, event.Data) + msg, err := ws.NewNotification(action, payload) if err != nil { b.logger.Error("failed to build websocket notification", zap.String("action", action), zap.Error(err)) return nil @@ -70,3 +66,44 @@ func (b *UserEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action } b.subscriptions = append(b.subscriptions, sub) } + +// routingUserIDAndPayload resolves the user to route event data to, and the +// payload to send them with "user_id" stripped. +// +// event.Data arrives in one of two shapes depending on transport: +// - MemoryEventBus delivers the original Go value in-process, so a +// struct's exported fields are already there. +// - a NATS-backed EventBus marshals every event before publishing and +// JSON-decodes it into an untyped Data field on arrival +// (internal/events/bus/nats.go), so a subscriber always sees a +// map[string]interface{} — an unexported struct field is invisible to +// that marshal and is silently gone by the time it gets here. +// +// Both shapes are normalized to a map (round-tripping a struct through JSON +// first) so every event type is read and routed the same way regardless of +// transport, and one strip covers both: "user_id" is a routing-only concern +// the client already knows (it's their own id) and must never appear in the +// message they actually receive. +func routingUserIDAndPayload(data interface{}) (userID string, payload map[string]interface{}) { + raw, ok := data.(map[string]interface{}) + if !ok { + encoded, err := json.Marshal(data) + if err != nil { + return "", nil + } + if err := json.Unmarshal(encoded, &raw); err != nil { + return "", nil + } + } + userID, _ = raw["user_id"].(string) + if userID == "" { + return "", nil + } + payload = make(map[string]interface{}, len(raw)) + for k, v := range raw { + if k != "user_id" { + payload[k] = v + } + } + return userID, payload +} diff --git a/apps/backend/internal/gateway/websocket/user_notifications_test.go b/apps/backend/internal/gateway/websocket/user_notifications_test.go new file mode 100644 index 0000000000..cc95fdf1b1 --- /dev/null +++ b/apps/backend/internal/gateway/websocket/user_notifications_test.go @@ -0,0 +1,169 @@ +package websocket + +import ( + "context" + "encoding/json" + "testing" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// pluginUserStateTestEvent mirrors internal/plugins' pluginUserStateUpdatedEvent +// (a plain struct with an exported UserID, json:"user_id") without a +// plugins package import (avoiding a gateway/websocket -> plugins dependency +// this package doesn't otherwise have). +type pluginUserStateTestEvent struct { + UserID string `json:"user_id"` + PluginID string `json:"pluginId"` + Key string `json:"key"` +} + +func testLoggerForUserNotifications(t *testing.T) *logger.Logger { + t.Helper() + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + return log +} + +// TestRegisterUserNotifications_PluginUserStateReachesOnlyWriter pins AC24: +// a plugin.user-state.updated bus event reaches only the writing user's WS +// connections; a second, differently-identified user's client receives +// nothing. +func TestRegisterUserNotifications_PluginUserStateReachesOnlyWriter(t *testing.T) { + h := newTestHub(t) + writer := newTestClient("writer-conn") + other := newTestClient("other-conn") + registerTestClient(h, writer) + registerTestClient(h, other) + h.SubscribeToUser(writer, "user_1") + h.SubscribeToUser(other, "user_2") + + eventBus := bus.NewMemoryEventBus(testLoggerForUserNotifications(t)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + RegisterUserNotifications(ctx, eventBus, h, testLoggerForUserNotifications(t)) + + payload := pluginUserStateTestEvent{UserID: "user_1", PluginID: "kandev-plugin-notes", Key: "note"} + if err := eventBus.Publish(ctx, events.PluginUserStateUpdated, + bus.NewEvent(events.PluginUserStateUpdated, "test", payload)); err != nil { + t.Fatalf("publish: %v", err) + } + + if !clientReceived(writer) { + t.Fatal("expected the writing user's client to receive the notification") + } + if clientReceived(other) { + t.Fatal("expected a different user's client to receive nothing") + } +} + +// TestRegisterUserNotifications_PluginUserStatePayloadShape pins that the WS +// notification payload carries exactly the plugin/scope/key fields (AC24) — +// no user_id leaks into the message the client receives. +func TestRegisterUserNotifications_PluginUserStatePayloadShape(t *testing.T) { + h := newTestHub(t) + writer := newTestClient("writer-conn") + registerTestClient(h, writer) + h.SubscribeToUser(writer, "user_1") + + eventBus := bus.NewMemoryEventBus(testLoggerForUserNotifications(t)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + RegisterUserNotifications(ctx, eventBus, h, testLoggerForUserNotifications(t)) + + payload := pluginUserStateTestEvent{UserID: "user_1", PluginID: "kandev-plugin-notes", Key: "note"} + if err := eventBus.Publish(ctx, events.PluginUserStateUpdated, + bus.NewEvent(events.PluginUserStateUpdated, "test", payload)); err != nil { + t.Fatalf("publish: %v", err) + } + + select { + case raw := <-writer.send: + var msg ws.Message + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("unmarshal envelope: %v", err) + } + if msg.Action != "plugin.user-state.updated" { + t.Fatalf("Action = %q, want %q", msg.Action, "plugin.user-state.updated") + } + var got map[string]any + if err := json.Unmarshal(msg.Payload, &got); err != nil { + t.Fatalf("unmarshal payload: %v", err) + } + if got["pluginId"] != "kandev-plugin-notes" || got["key"] != "note" { + t.Fatalf("payload = %+v, missing expected fields", got) + } + if _, leaked := got["user_id"]; leaked { + t.Fatalf("payload leaked user_id: %+v", got) + } + default: + t.Fatal("expected a message on writer.send") + } +} + +// TestRegisterUserNotifications_PluginUserStateRoutesUnderNATSTransport is +// the regression test for the NATS routing gap found in review: a +// NATS-backed EventBus marshals every event before publishing and +// JSON-decodes it into an untyped Data field on arrival +// (internal/events/bus/nats.go's createMsgHandler does +// json.Unmarshal(msg.Data, &event)), so a subscriber always sees +// event.Data as a map[string]interface{} — never the original Go struct. +// This reproduces exactly that shape (marshal the real event struct, then +// unmarshal into a bus.Event the same way nats.go does) without needing a +// live NATS server, and proves routing plus the no-leak-to-the-browser +// contract both still hold once event.Data arrives as a map instead of a +// struct. +func TestRegisterUserNotifications_PluginUserStateRoutesUnderNATSTransport(t *testing.T) { + h := newTestHub(t) + writer := newTestClient("writer-conn") + registerTestClient(h, writer) + h.SubscribeToUser(writer, "user_1") + + eventBus := bus.NewMemoryEventBus(testLoggerForUserNotifications(t)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + RegisterUserNotifications(ctx, eventBus, h, testLoggerForUserNotifications(t)) + + published := bus.NewEvent(events.PluginUserStateUpdated, "test", + pluginUserStateTestEvent{UserID: "user_1", PluginID: "kandev-plugin-notes", Key: "note"}) + wire, err := json.Marshal(published) + if err != nil { + t.Fatalf("marshal (simulating what a NATS publisher sends over the wire): %v", err) + } + var arrived bus.Event + if err := json.Unmarshal(wire, &arrived); err != nil { + t.Fatalf("unmarshal (simulating nats.go's createMsgHandler on receipt): %v", err) + } + if _, ok := arrived.Data.(map[string]interface{}); !ok { + t.Fatalf("arrived.Data = %T, want map[string]interface{} (the whole point of this test)", arrived.Data) + } + + if err := eventBus.Publish(ctx, events.PluginUserStateUpdated, &arrived); err != nil { + t.Fatalf("publish: %v", err) + } + + select { + case raw := <-writer.send: + var msg ws.Message + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("unmarshal envelope: %v", err) + } + var got map[string]any + if err := json.Unmarshal(msg.Payload, &got); err != nil { + t.Fatalf("unmarshal payload: %v", err) + } + if got["pluginId"] != "kandev-plugin-notes" || got["key"] != "note" { + t.Fatalf("payload = %+v, missing expected fields", got) + } + if _, leaked := got["user_id"]; leaked { + t.Fatalf("payload leaked user_id: %+v", got) + } + default: + t.Fatal("expected the writing user's client to receive the notification even when event.Data arrives as a map (NATS transport)") + } +} diff --git a/apps/backend/internal/plugins/handlers.go b/apps/backend/internal/plugins/handlers.go index c7c5a2940b..23a35b50e3 100644 --- a/apps/backend/internal/plugins/handlers.go +++ b/apps/backend/internal/plugins/handlers.go @@ -63,6 +63,10 @@ func RegisterRoutes(router *gin.Engine, svc *Service, _ Deliverer, log *logger.L api.GET("/:id/bundle", ctrl.bundle) api.GET("/:id/ui/*path", ctrl.ui) + // Registered before the /:id/webhooks/:key wildcard for the same reason + // /settings is registered before /:id above: some gin/httprouter tree + // versions reject a static-ish sibling added after an existing wildcard. + registerUserStateRoutes(api, ctrl) api.POST("/:id/webhooks/:key", ctrl.webhook) api.GET("/:id/webhooks/:key", ctrl.webhook) } diff --git a/apps/backend/internal/plugins/manifest/manifest.go b/apps/backend/internal/plugins/manifest/manifest.go index c53b85642d..ab9ad71279 100644 --- a/apps/backend/internal/plugins/manifest/manifest.go +++ b/apps/backend/internal/plugins/manifest/manifest.go @@ -91,6 +91,13 @@ type Capabilities struct { // can log a visitor in as any (or a new) user — so it should only be // granted to plugins an operator explicitly trusts. Auth bool `yaml:"auth,omitempty" json:"auth,omitempty"` + // UserState gates the authenticated, browser-reachable per-user storage + // routes (PUT/GET/DELETE /api/plugins/:id/user-state/...). Unlike State + // (plugin_state, written by the plugin's own gRPC-connected backend), + // this lets a UI-only plugin bundle persist data scoped to the calling + // browser session's user with no Go backend of its own (Approach D1, + // docs/decisions/2026-08-01-per-user-plugin-storage.md). + UserState bool `yaml:"user_state,omitempty" json:"user_state,omitempty"` } // AuthProvider is a login option a plugin contributes to the pre-auth login diff --git a/apps/backend/internal/plugins/manifest/manifest_test.go b/apps/backend/internal/plugins/manifest/manifest_test.go index a2f831ba3f..539dc068e2 100644 --- a/apps/backend/internal/plugins/manifest/manifest_test.go +++ b/apps/backend/internal/plugins/manifest/manifest_test.go @@ -57,6 +57,30 @@ func TestParse_ValidManifestParsesID(t *testing.T) { } } +// TestParse_UserStateCapabilityRoundTrips pins capabilities.user_state +// (Approach D1 / AC17): a plugin declaring host-provided per-user storage +// parses with Capabilities.UserState set. +func TestParse_UserStateCapabilityRoundTrips(t *testing.T) { + yaml := strings.Replace(validManifestYAML, "secrets: true", "secrets: true\n user_state: true", 1) + m, err := Parse([]byte(yaml)) + if err != nil { + t.Fatalf("Parse() unexpected error: %v", err) + } + if !m.Capabilities.UserState { + t.Fatalf("Capabilities.UserState = false, want true") + } +} + +// TestParse_UserStateCapabilityDefaultsFalse pins that omitting +// capabilities.user_state defaults to false — a plugin must opt in before +// the per-user storage routes accept its writes (AC17). +func TestParse_UserStateCapabilityDefaultsFalse(t *testing.T) { + m := validManifest(t) + if m.Capabilities.UserState { + t.Fatalf("Capabilities.UserState = true, want false (not declared in fixture)") + } +} + func TestValidate_ValidManifestPasses(t *testing.T) { m, err := Parse([]byte(validManifestYAML)) if err != nil { diff --git a/apps/backend/internal/plugins/provider.go b/apps/backend/internal/plugins/provider.go index e25673e51e..b245dc9010 100644 --- a/apps/backend/internal/plugins/provider.go +++ b/apps/backend/internal/plugins/provider.go @@ -61,6 +61,11 @@ func Provide(cfg *config.Config, dbPool *db.Pool, secrets SecretVault, eventBus return nil, nil, fmt.Errorf("plugins: init state store: %w", err) } + userStateStore, err := state.NewUserStore(dbPool) + if err != nil { + return nil, nil, fmt.Errorf("plugins: init user state store: %w", err) + } + registry := NewRegistry() if err := registry.Load(pluginStore); err != nil { return nil, nil, fmt.Errorf("plugins: load registry: %w", err) @@ -68,6 +73,7 @@ func Provide(cfg *config.Config, dbPool *db.Pool, secrets SecretVault, eventBus svc := NewService(pluginStore, registry, eventBus, log) svc.SetState(stateStore) + svc.SetUserState(userStateStore) svc.SetSecrets(secrets) svc.SetPluginsDir(dir) diff --git a/apps/backend/internal/plugins/service.go b/apps/backend/internal/plugins/service.go index e208e0a5a3..b27c1803b4 100644 --- a/apps/backend/internal/plugins/service.go +++ b/apps/backend/internal/plugins/service.go @@ -32,6 +32,10 @@ const maxDownloadSize = 100 << 20 // download. const downloadTimeout = 60 * time.Second +type userStateCleanupStore interface { + DeleteAllForPlugin(context.Context, string) error +} + // Service is the core plugin service: install/uninstall, the in-memory // Registry, the lifecycle state machine, and the runtime.Manager wiring // that spawns/supervises each plugin's subprocess. @@ -71,12 +75,14 @@ type Service struct { // cannot deadlock against it. lifecycleLocks *keyedMutex - pluginsDir string - store store.Store - registry *Registry - state *state.Store - eventBus bus.EventBus - log *logger.Logger + pluginsDir string + store store.Store + registry *Registry + state *state.Store + userState *state.UserStore + userStateCleanup userStateCleanupStore + eventBus bus.EventBus + log *logger.Logger deliverer Deliverer runtime PluginRuntime @@ -204,6 +210,32 @@ func (s *Service) StateStore() *state.Store { return s.state } +// SetUserState wires the already-constructed plugin_user_state store. Provide +// calls this; also exposed for tests that build a Service without going +// through Provide. See UserState's doc comment for how it differs from +// StateStore. +func (s *Service) SetUserState(st *state.UserStore) { + s.userState = st + s.userStateCleanup = st +} + +// setUserStateCleanupStore replaces the narrow uninstall-cleanup seam. The +// concrete UserStore remains the handler-facing store; this seam lets tests +// exercise fail-closed uninstall behavior without weakening that accessor. +func (s *Service) setUserStateCleanupStore(cleanup userStateCleanupStore) { + s.userStateCleanup = cleanup +} + +// UserState returns the plugin_user_state store Provide constructed, for the +// authenticated per-user storage HTTP routes (user_state_handlers.go). +// Unlike StateStore (plugin_state, written only by a plugin's own +// gRPC-connected backend via the Host RPC), this store is reachable directly +// from an authenticated browser request — every read/write is scoped to the +// calling user (Approach D1, docs/decisions/2026-08-01-per-user-plugin-storage.md). +func (s *Service) UserState() *state.UserStore { + return s.userState +} + // SetSecrets wires the secret vault Provide was constructed with. func (s *Service) SetSecrets(v SecretVault) { s.secrets = v @@ -858,23 +890,23 @@ func validateInstallURL(raw string) error { return nil } -// Uninstall stops id's process (if running), purges its vault namespace, -// removes its extracted package tree from disk, deletes its record from both -// the store and the in-memory registry, and deletes every plugin_state row -// scoped to id (best-effort — a failure there is logged but does not fail -// the overall Uninstall, since the package/record are already gone by that -// point), then notifies the attached Deliverer. Clearing plugin_state and -// the vault namespace matters so a plugin reinstalled under the same id (or -// an id later reused by a different plugin) never silently inherits stale -// state or secrets. +// Uninstall stops id's process (if running), purges its vault namespace and +// every plugin_user_state row, removes its extracted package tree from disk, +// deletes its record from both the store and the in-memory registry, and +// deletes every plugin_state row scoped to id (best-effort — a failure there +// is logged but does not fail the overall Uninstall, since the package/record +// are already gone by that point), then notifies the attached Deliverer. +// Clearing plugin_state, plugin_user_state, and the vault namespace matters so +// a plugin reinstalled under the same id (or an id later reused by a different +// plugin) never silently inherits stale state or secrets. // // Ordering is deliberate: the process is stopped FIRST so the plugin can no // longer race the cleanup by writing a fresh secret (SetSecret) between the // vault list and the deletes; then the vault namespace is purged and any // failure aborts the uninstall — nothing destructive (package/record -// removal) has happened yet, so the operator simply retries (Stop and the -// vault deletes are both idempotent). A failed uninstall therefore leaves -// the plugin stopped-but-installed, resolved by a retry. +// removal) has happened yet, so the operator simply retries (Stop, the vault +// deletes, and the user-state purge are all idempotent). A failed uninstall +// therefore leaves the plugin stopped-but-installed, resolved by a retry. func (s *Service) Uninstall(ctx context.Context, id string) error { lock := s.lifecycleLocks.lockFor(id) lock.Lock() @@ -888,19 +920,13 @@ func (s *Service) Uninstall(ctx context.Context, id string) error { s.runtime.Stop(id) } if err := s.deletePluginSecrets(ctx, id); err != nil { - // The process is stopped but nothing else was removed. If it had been - // running, its persisted status still says active — reconcile it to - // error and notify observers so it isn't reported as running while no - // process is, before returning the retryable failure. - if wasRunning { - if setErr := s.SetStatus(id, StatusError); setErr != nil { - s.log.Warn("plugins: could not mark plugin errored after an aborted uninstall", - zap.String("plugin_id", id), zap.Error(setErr)) - } - s.notifyDeliverer() - } + s.reconcileAbortedUninstall(id, wasRunning) return fmt.Errorf("plugins: uninstall aborted, could not purge plugin secrets: %w", err) } + if err := s.deletePluginUserState(ctx, id); err != nil { + s.reconcileAbortedUninstall(id, wasRunning) + return fmt.Errorf("plugins: uninstall aborted, could not purge plugin user state: %w", err) + } if err := pkgtar.Remove(s.pluginsDir, id); err != nil { return fmt.Errorf("plugins: remove installed package: %w", err) } @@ -913,6 +939,17 @@ func (s *Service) Uninstall(ctx context.Context, id string) error { return nil } +func (s *Service) reconcileAbortedUninstall(id string, wasRunning bool) { + if !wasRunning { + return + } + if setErr := s.SetStatus(id, StatusError); setErr != nil { + s.log.Warn("plugins: could not mark plugin errored after an aborted uninstall", + zap.String("plugin_id", id), zap.Error(setErr)) + } + s.notifyDeliverer() +} + // deletePluginSecrets removes every vault entry in id's namespace // ("plugin::..." — both SetSecret-owned and config-backed entries), so // a reinstall under the same id never inherits stale secrets. Unlike @@ -954,6 +991,16 @@ func (s *Service) deletePluginState(id string) { } } +// deletePluginUserState removes every plugin_user_state row for id, across +// every user (AC20). A nil cleanup store is a silent no-op for narrowly +// constructed tests where per-user storage could never have been written. +func (s *Service) deletePluginUserState(ctx context.Context, id string) error { + if s.userStateCleanup == nil { + return nil + } + return s.userStateCleanup.DeleteAllForPlugin(ctx, id) +} + // Enable transitions id to StatusActive, spawning its process first if it // is not already running. Idempotent: a no-op (nil error) if id is already // active. diff --git a/apps/backend/internal/plugins/service_test.go b/apps/backend/internal/plugins/service_test.go index 2a6a485395..1134c8d7be 100644 --- a/apps/backend/internal/plugins/service_test.go +++ b/apps/backend/internal/plugins/service_test.go @@ -54,6 +54,20 @@ type fakeRuntime struct { blockProceed chan struct{} } +type fakeUserStateCleanup struct { + deleteErr error + delete func(context.Context, string) error + calls int +} + +func (f *fakeUserStateCleanup) DeleteAllForPlugin(ctx context.Context, pluginID string) error { + f.calls++ + if f.deleteErr != nil { + return f.deleteErr + } + return f.delete(ctx, pluginID) +} + func newFakeRuntime() *fakeRuntime { return &fakeRuntime{ running: map[string]bool{}, @@ -406,6 +420,98 @@ func TestServiceUninstallDeletesPluginState(t *testing.T) { } } +// TestServiceUninstallDeletesPluginUserStateForEveryUser pins AC20: the +// per-user counterpart of TestServiceUninstallDeletesPluginState — uninstall +// must purge plugin_user_state rows for every user who wrote one, not just +// whichever user happened to trigger the uninstall. +func TestServiceUninstallDeletesPluginUserStateForEveryUser(t *testing.T) { + svc, _, _ := newTestService(t) + svc.SetUserState(newTestUserStateStore(t)) + installTestPlugin(t, svc, "kandev-plugin-notes") + + ctx := context.Background() + if _, err := svc.UserState().Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_1", "note", json.RawMessage(`"a"`), nil); err != nil { + t.Fatalf("seed user_1: %v", err) + } + if _, err := svc.UserState().Set(ctx, "kandev-plugin-notes", "user_2", "task", "task_1", "note", json.RawMessage(`"b"`), nil); err != nil { + t.Fatalf("seed user_2: %v", err) + } + + if err := svc.Uninstall(context.Background(), "kandev-plugin-notes"); err != nil { + t.Fatalf("Uninstall() unexpected error: %v", err) + } + + for _, userID := range []string{"user_1", "user_2"} { + entries, err := svc.UserState().List(ctx, "kandev-plugin-notes", userID, "task", "task_1") + if err != nil { + t.Fatalf("List(%s) after Uninstall(): %v", userID, err) + } + if len(entries) != 0 { + t.Fatalf("plugin_user_state entries for %s after Uninstall() = %d, want 0", userID, len(entries)) + } + } +} + +func TestServiceUninstallFailsClosedWhenUserStateCleanupFails(t *testing.T) { + svc, fsStore, rt := newTestService(t) + svc.SetUserState(newTestUserStateStore(t)) + rec := installTestPlugin(t, svc, "kandev-plugin-notes") + ctx := context.Background() + if _, err := svc.UserState().Set(ctx, rec.ID, "user_1", "task", "task_1", "note", json.RawMessage(`"a"`), nil); err != nil { + t.Fatalf("seed user state: %v", err) + } + + cleanupErr := errors.New("user state database unavailable") + cleanup := &fakeUserStateCleanup{ + deleteErr: cleanupErr, + delete: svc.UserState().DeleteAllForPlugin, + } + svc.setUserStateCleanupStore(cleanup) + deliverer := &fakeDeliverer{} + svc.SetDeliverer(deliverer) + + err := svc.Uninstall(ctx, rec.ID) + if err == nil || !strings.Contains(err.Error(), cleanupErr.Error()) { + t.Fatalf("Uninstall() error = %v, want user-state cleanup failure", err) + } + if cleanup.calls != 1 { + t.Fatalf("DeleteAllForPlugin calls after failed uninstall = %d, want 1", cleanup.calls) + } + if !rt.stopped(rec.ID) { + t.Fatal("Uninstall() did not stop the runtime before cleanup failure") + } + if _, err := svc.Get(rec.ID); err != nil { + t.Fatalf("Get() after failed uninstall: %v, want installed record", err) + } + if _, err := fsStore.Get(rec.ID); err != nil { + t.Fatalf("store.Get() after failed uninstall: %v, want installed record", err) + } + if _, err := os.Stat(rec.InstallPath); err != nil { + t.Fatalf("installed package after failed uninstall: %v", err) + } + if deliverer.refreshCount != 1 { + t.Fatalf("deliverer refreshes after failed uninstall = %d, want stopped-state reconciliation only", deliverer.refreshCount) + } + + cleanup.deleteErr = nil + if err := svc.Uninstall(ctx, rec.ID); err != nil { + t.Fatalf("retry Uninstall() error: %v", err) + } + if cleanup.calls != 2 { + t.Fatalf("DeleteAllForPlugin calls after retry = %d, want 2", cleanup.calls) + } + if _, err := svc.Get(rec.ID); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("Get() after successful retry = %v, want store.ErrNotFound", err) + } + entries, err := svc.UserState().List(ctx, rec.ID, "user_1", "task", "task_1") + if err != nil { + t.Fatalf("List() after successful retry: %v", err) + } + if len(entries) != 0 { + t.Fatalf("user state rows after successful retry = %d, want 0", len(entries)) + } +} + func TestServiceUninstallMissingReturnsNotFound(t *testing.T) { svc, _, _ := newTestService(t) if err := svc.Uninstall(context.Background(), "missing"); !errors.Is(err, store.ErrNotFound) { diff --git a/apps/backend/internal/plugins/state/goleak_test.go b/apps/backend/internal/plugins/state/goleak_test.go new file mode 100644 index 0000000000..a8ef34c290 --- /dev/null +++ b/apps/backend/internal/plugins/state/goleak_test.go @@ -0,0 +1,15 @@ +package state + +import ( + "testing" + + "go.uber.org/goleak" +) + +// TestMain asserts that no goroutines from this package outlive the test +// process. TestUserStorePostgresConditionalWriteIsRaceFreeUnderConcurrency +// races writers across goroutines it must join before returning — goleak +// catches a regression where one is left running past the test. +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m) +} diff --git a/apps/backend/internal/plugins/state/user_store.go b/apps/backend/internal/plugins/state/user_store.go new file mode 100644 index 0000000000..2a2cf46af1 --- /dev/null +++ b/apps/backend/internal/plugins/state/user_store.go @@ -0,0 +1,228 @@ +package state + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jmoiron/sqlx" + + "github.com/kandev/kandev/internal/db" +) + +// ErrConflict is returned by UserStore.Set when the caller supplies +// ifUnmodifiedSince and the stored row was modified after that time +// (Approach H1's optimistic-concurrency guard, AC28). The stored value is +// left unchanged. +var ErrConflict = errors.New("plugin user state: conflict") + +// userStateTimeLayout is RFC3339 with a fixed 9-digit fractional part. +// time.RFC3339Nano trims trailing zeros — a value with a zero nanosecond +// component formats with no fractional part at all (e.g. +// "2026-08-01T10:00:00Z"), which sorts *after* a same-second value that does +// have a fraction (e.g. "2026-08-01T10:00:00.000000001Z", since 'Z' > '.' +// byte-wise). updated_at is a TEXT column compared lexicographically in the +// ifUnmodifiedSince WHERE clause, so that reordering would let a stale +// conditional write through and silently discard another surface's edit — +// exactly what AC28 exists to prevent. A fixed width keeps lexicographic +// order equal to chronological order for every value. +const userStateTimeLayout = "2006-01-02T15:04:05.000000000Z07:00" + +// UserStateEntry is a single row returned by UserStore.List. +type UserStateEntry struct { + Key string `db:"state_key" json:"key"` + Value json.RawMessage `db:"value_json" json:"value"` + UpdatedAt time.Time `db:"updated_at" json:"updatedAt"` +} + +// UserStore persists per-user, browser-authenticated plugin state in the +// plugin_user_state table — the counterpart to Store (plugin_state), which +// has no user dimension and is only ever written by a plugin's own +// gRPC-connected backend. A row here is keyed by +// (plugin_id, user_id, scope, scope_id, state_key); no plugin or user can +// read another's row (docs/decisions/2026-08-01-per-user-plugin-storage.md). +type UserStore struct { + db *sqlx.DB + ro *sqlx.DB +} + +// NewUserStore creates a UserStore and initializes the plugin_user_state +// schema if needed. +func NewUserStore(pool *db.Pool) (*UserStore, error) { + s := &UserStore{db: pool.Writer(), ro: pool.Reader()} + if err := s.initSchema(); err != nil { + return nil, fmt.Errorf("plugin user state schema init: %w", err) + } + return s, nil +} + +// initSchema creates the plugin_user_state table. scope_id follows Store's +// convention (NOT NULL DEFAULT ”) rather than a nullable column, for the +// same reason: SQLite's UNIQUE index treats every NULL as distinct, so a +// nullable scope_id would let ON CONFLICT silently miss conflicts between +// instance-scoped rows. +func (s *UserStore) initSchema() error { + _, err := s.db.Exec(` + CREATE TABLE IF NOT EXISTS plugin_user_state ( + id TEXT PRIMARY KEY, + plugin_id TEXT NOT NULL, + user_id TEXT NOT NULL, + scope TEXT NOT NULL DEFAULT 'instance', + scope_id TEXT NOT NULL DEFAULT '', + state_key TEXT NOT NULL, + value_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE (plugin_id, user_id, scope, scope_id, state_key) + ); + `) + return err +} + +// Get returns the value and updated_at stored for the given +// plugin/user/scope/scopeID/key. found is false if no row exists. +func (s *UserStore) Get( + ctx context.Context, pluginID, userID, scope, scopeID, key string, +) (json.RawMessage, time.Time, bool, error) { + var raw, updatedAtStr string + err := s.ro.QueryRowContext(ctx, s.ro.Rebind(` + SELECT value_json, updated_at FROM plugin_user_state + WHERE plugin_id = ? AND user_id = ? AND scope = ? AND scope_id = ? AND state_key = ? + `), pluginID, userID, scope, scopeID, key).Scan(&raw, &updatedAtStr) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, time.Time{}, false, nil + } + return nil, time.Time{}, false, err + } + updatedAt, err := time.Parse(time.RFC3339, updatedAtStr) + if err != nil { + return nil, time.Time{}, false, fmt.Errorf("parse updated_at for key %q: %w", key, err) + } + return json.RawMessage(raw), updatedAt, true, nil +} + +// Set upserts the value for the given plugin/user/scope/scopeID/key, setting +// updated_at to the current time (returned), stored via userStateTimeLayout. +// +// The layout must keep sub-second precision: ifUnmodifiedSince is compared +// against the stored value, so storing at whole-second resolution would make +// every pair of writes inside the same second indistinguishable and let a +// stale conditional write through (silently discarding the other surface's +// edit — exactly what H1/AC28 exists to prevent). Reads still parse with +// time.RFC3339, which accepts an optional fractional part, so rows written +// before this layout change (RFC3339Nano's variable-width fraction) +// continue to load; only new writes get the fixed width. +// +// When ifUnmodifiedSince is non-nil, the conflict check is folded into the +// upsert's ON CONFLICT ... DO UPDATE ... WHERE clause rather than a separate +// SELECT-then-write: a plain read-then-act check (even inside a transaction) +// only blocks concurrent writers under serializable isolation. This codebase +// also runs against PostgreSQL, whose default READ COMMITTED isolation lets +// two concurrent transactions both read the same "not yet conflicting" row +// and both proceed to write, each silently clobbering the other. Folding the +// timestamp comparison into the UPDATE's WHERE clause makes the check and +// the write one atomic, row-locking operation regardless of isolation level: +// if the WHERE fails (a newer row won the race), neither the UPDATE nor the +// INSERT-on-no-conflict branch applies, RowsAffected is 0, and the caller +// gets ErrConflict with the stored value left untouched. A nil +// ifUnmodifiedSince, or no existing row (nothing for ON CONFLICT to match), +// always writes (today's unconditional last-write-wins). +func (s *UserStore) Set( + ctx context.Context, pluginID, userID, scope, scopeID, key string, + value json.RawMessage, ifUnmodifiedSince *time.Time, +) (time.Time, error) { + now := time.Now().UTC() + query := ` + INSERT INTO plugin_user_state (id, plugin_id, user_id, scope, scope_id, state_key, value_json, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(plugin_id, user_id, scope, scope_id, state_key) + DO UPDATE SET value_json = excluded.value_json, updated_at = excluded.updated_at` + args := []any{ + uuid.New().String(), pluginID, userID, scope, scopeID, key, string(value), now.Format(userStateTimeLayout), + } + if ifUnmodifiedSince != nil { + query += ` WHERE plugin_user_state.updated_at <= ?` + args = append(args, ifUnmodifiedSince.UTC().Format(userStateTimeLayout)) + } + + res, err := s.db.ExecContext(ctx, s.db.Rebind(query), args...) + if err != nil { + return time.Time{}, err + } + if ifUnmodifiedSince != nil { + affected, err := res.RowsAffected() + if err != nil { + return time.Time{}, err + } + if affected == 0 { + return time.Time{}, ErrConflict + } + } + return now, nil +} + +// Delete removes the row for the given plugin/user/scope/scopeID/key. It is +// not an error if no matching row exists. +func (s *UserStore) Delete(ctx context.Context, pluginID, userID, scope, scopeID, key string) error { + _, err := s.db.ExecContext(ctx, s.db.Rebind(` + DELETE FROM plugin_user_state + WHERE plugin_id = ? AND user_id = ? AND scope = ? AND scope_id = ? AND state_key = ? + `), pluginID, userID, scope, scopeID, key) + return err +} + +// List returns every state entry for the given plugin/user/scope/scopeID, +// ordered by key (AC27), mirroring Store.List's ORDER BY state_key. +func (s *UserStore) List(ctx context.Context, pluginID, userID, scope, scopeID string) ([]UserStateEntry, error) { + rows, err := s.ro.QueryContext(ctx, s.ro.Rebind(` + SELECT state_key, value_json, updated_at FROM plugin_user_state + WHERE plugin_id = ? AND user_id = ? AND scope = ? AND scope_id = ? + ORDER BY state_key + `), pluginID, userID, scope, scopeID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + var entries []UserStateEntry + for rows.Next() { + var key, raw, updatedAtStr string + if err := rows.Scan(&key, &raw, &updatedAtStr); err != nil { + return nil, err + } + updatedAt, err := time.Parse(time.RFC3339, updatedAtStr) + if err != nil { + return nil, fmt.Errorf("parse updated_at for key %q: %w", key, err) + } + entries = append(entries, UserStateEntry{Key: key, Value: json.RawMessage(raw), UpdatedAt: updatedAt}) + } + if err := rows.Err(); err != nil { + return nil, err + } + return entries, nil +} + +// DeleteAllForPlugin removes every plugin_user_state row for pluginID, +// across every user, scope, and scope_id. Called by Service.Uninstall (AC20) +// so a reinstalled or id-reused plugin never inherits stale per-user state. +func (s *UserStore) DeleteAllForPlugin(ctx context.Context, pluginID string) error { + _, err := s.db.ExecContext(ctx, s.db.Rebind(` + DELETE FROM plugin_user_state WHERE plugin_id = ? + `), pluginID) + return err +} + +// DeleteAllForUser removes every plugin_user_state row for userID, across +// every plugin, scope, and scope_id. Not currently wired into a user +// deletion path — no such cascade hook exists yet in this codebase (R6) — +// but exposed so one can call it directly when that hook is added. +func (s *UserStore) DeleteAllForUser(ctx context.Context, userID string) error { + _, err := s.db.ExecContext(ctx, s.db.Rebind(` + DELETE FROM plugin_user_state WHERE user_id = ? + `), userID) + return err +} diff --git a/apps/backend/internal/plugins/state/user_store_test.go b/apps/backend/internal/plugins/state/user_store_test.go new file mode 100644 index 0000000000..63d8c33d27 --- /dev/null +++ b/apps/backend/internal/plugins/state/user_store_test.go @@ -0,0 +1,564 @@ +package state + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jmoiron/sqlx" + + "github.com/kandev/kandev/internal/db" + "github.com/kandev/kandev/internal/testutil" +) + +func TestUserStoreSetThenGetReturnsStoredValue(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + value := json.RawMessage(`{"text":"hello"}`) + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", value, nil); err != nil { + t.Fatalf("set: %v", err) + } + + got, _, found, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("get: %v", err) + } + if !found { + t.Fatalf("expected found = true") + } + if string(got) != string(value) { + t.Fatalf("got %q, want %q", got, value) + } +} + +func TestUserStoreGetMissingReturnsNotFound(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + got, _, found, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "missing") + if err != nil { + t.Fatalf("get: %v", err) + } + if found || got != nil { + t.Fatalf("got (%q, %v), want (nil, false)", got, found) + } +} + +// TestUserStoreIsolatesByUser pins the core trust boundary of +// plugin_user_state: two different users writing the same +// plugin/scope/scopeID/key must never see each other's value (a different +// user's Get returns not-found, matching AC15's cross-user 404). +func TestUserStoreIsolatesByUser(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"alice's note"`), nil); err != nil { + t.Fatalf("set user_1: %v", err) + } + + got, _, found, err := store.Get(ctx, "kandev-plugin-notes", "user_2", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("get user_2: %v", err) + } + if found { + t.Fatalf("expected user_2 to not see user_1's note, got %q", got) + } + + // user_1 still sees their own value. + got, _, found, err = store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("get user_1: %v", err) + } + if !found || string(got) != `"alice's note"` { + t.Fatalf("got (%q, %v), want (\"alice's note\", true)", got, found) + } +} + +func TestUserStoreSetUpsertsOnRepeatedWrite(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v1"`), nil); err != nil { + t.Fatalf("first set: %v", err) + } + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v2"`), nil); err != nil { + t.Fatalf("second set: %v", err) + } + + got, _, found, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("get: %v", err) + } + if !found || string(got) != `"v2"` { + t.Fatalf("got (%q, %v), want (\"v2\", true) (upsert should overwrite)", got, found) + } + + entries, err := store.List(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(entries) != 1 { + t.Fatalf("expected exactly 1 entry after repeated Set, got %d: %+v", len(entries), entries) + } +} + +// TestUserStoreDistinctKeysAreIndependent pins AC27: writes to distinct keys +// never clobber one another. +func TestUserStoreDistinctKeysAreIndependent(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "a", json.RawMessage(`1`), nil); err != nil { + t.Fatalf("set a: %v", err) + } + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "b", json.RawMessage(`2`), nil); err != nil { + t.Fatalf("set b: %v", err) + } + + gotA, _, _, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "a") + if err != nil { + t.Fatalf("get a: %v", err) + } + gotB, _, _, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "b") + if err != nil { + t.Fatalf("get b: %v", err) + } + if string(gotA) != "1" || string(gotB) != "2" { + t.Fatalf("got a=%q b=%q, want a=1 b=2 (independent)", gotA, gotB) + } +} + +// TestUserStoreListReturnsOrderedByKey pins AC27's ordering contract, mirroring +// Store.List's ORDER BY state_key. +func TestUserStoreListReturnsOrderedByKey(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + for _, key := range []string{"zeta", "alpha", "mu"} { + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", key, json.RawMessage(`1`), nil); err != nil { + t.Fatalf("set %s: %v", key, err) + } + } + + entries, err := store.List(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(entries) != 3 { + t.Fatalf("expected 3 entries, got %d", len(entries)) + } + want := []string{"alpha", "mu", "zeta"} + for i, e := range entries { + if e.Key != want[i] { + t.Fatalf("entries[%d].Key = %q, want %q (order: %+v)", i, e.Key, want[i], entries) + } + } +} + +func TestUserStoreListDoesNotLeakAcrossUsers(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"a"`), nil); err != nil { + t.Fatalf("set user_1: %v", err) + } + + entries, err := store.List(ctx, "kandev-plugin-notes", "user_2", "task", "task_xyz") + if err != nil { + t.Fatalf("list user_2: %v", err) + } + if len(entries) != 0 { + t.Fatalf("expected no entries visible to user_2, got %+v", entries) + } +} + +func TestUserStoreDeleteRemovesEntry(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"a"`), nil); err != nil { + t.Fatalf("set: %v", err) + } + if err := store.Delete(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note"); err != nil { + t.Fatalf("delete: %v", err) + } + + _, _, found, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("get after delete: %v", err) + } + if found { + t.Fatalf("expected not found after delete") + } +} + +func TestUserStoreDeleteMissingIsNotAnError(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if err := store.Delete(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "never_set"); err != nil { + t.Fatalf("delete missing: %v", err) + } +} + +// TestUserStoreDeleteAllForPluginPurgesEveryUser pins AC20: uninstalling a +// plugin must purge plugin_user_state rows for every user, not just one. +func TestUserStoreDeleteAllForPluginPurgesEveryUser(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"a"`), nil); err != nil { + t.Fatalf("set user_1: %v", err) + } + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_2", "task", "task_xyz", "note", json.RawMessage(`"b"`), nil); err != nil { + t.Fatalf("set user_2: %v", err) + } + if _, err := store.Set(ctx, "kandev-plugin-other", "user_1", "task", "task_xyz", "note", json.RawMessage(`"c"`), nil); err != nil { + t.Fatalf("set other plugin: %v", err) + } + + if err := store.DeleteAllForPlugin(ctx, "kandev-plugin-notes"); err != nil { + t.Fatalf("DeleteAllForPlugin: %v", err) + } + + if entries, err := store.List(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz"); err != nil || len(entries) != 0 { + t.Fatalf("user_1 entries after DeleteAllForPlugin = %v (err=%v), want none", entries, err) + } + if entries, err := store.List(ctx, "kandev-plugin-notes", "user_2", "task", "task_xyz"); err != nil || len(entries) != 0 { + t.Fatalf("user_2 entries after DeleteAllForPlugin = %v (err=%v), want none", entries, err) + } + // A different plugin's state is untouched. + if entries, err := store.List(ctx, "kandev-plugin-other", "user_1", "task", "task_xyz"); err != nil || len(entries) != 1 { + t.Fatalf("kandev-plugin-other entries after deleting kandev-plugin-notes = %v (err=%v), want 1 (untouched)", entries, err) + } +} + +// TestUserStoreDeleteAllForUserPurgesEveryPlugin covers the DeleteAllForUser +// helper (R6: a future user-deletion cascade hook). +func TestUserStoreDeleteAllForUserPurgesEveryPlugin(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"a"`), nil); err != nil { + t.Fatalf("set plugin-notes: %v", err) + } + if _, err := store.Set(ctx, "kandev-plugin-other", "user_1", "task", "task_xyz", "note", json.RawMessage(`"b"`), nil); err != nil { + t.Fatalf("set plugin-other: %v", err) + } + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_2", "task", "task_xyz", "note", json.RawMessage(`"c"`), nil); err != nil { + t.Fatalf("set user_2: %v", err) + } + + if err := store.DeleteAllForUser(ctx, "user_1"); err != nil { + t.Fatalf("DeleteAllForUser: %v", err) + } + + if entries, err := store.List(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz"); err != nil || len(entries) != 0 { + t.Fatalf("user_1/plugin-notes entries after DeleteAllForUser = %v (err=%v), want none", entries, err) + } + if entries, err := store.List(ctx, "kandev-plugin-other", "user_1", "task", "task_xyz"); err != nil || len(entries) != 0 { + t.Fatalf("user_1/plugin-other entries after DeleteAllForUser = %v (err=%v), want none", entries, err) + } + if entries, err := store.List(ctx, "kandev-plugin-notes", "user_2", "task", "task_xyz"); err != nil || len(entries) != 1 { + t.Fatalf("user_2 entries after deleting user_1 = %v (err=%v), want 1 (untouched)", entries, err) + } +} + +// TestUserStoreSetConditionalWriteConflict pins Approach H1 / AC28: a write +// carrying ifUnmodifiedSince older than the stored row's updated_at is +// rejected with ErrConflict and the stored value is left unchanged. +func TestUserStoreSetConditionalWriteConflict(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + firstUpdatedAt, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v1"`), nil) + if err != nil { + t.Fatalf("first set: %v", err) + } + + stale := firstUpdatedAt.Add(-time.Minute) + _, err = store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v2-conflict"`), &stale) + if !errors.Is(err, ErrConflict) { + t.Fatalf("expected ErrConflict, got %v", err) + } + + got, _, _, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("get: %v", err) + } + if string(got) != `"v1"` { + t.Fatalf("stored value changed after a conflicting write: got %q, want \"v1\"", got) + } +} + +// TestUserStoreSetConditionalWriteAcceptsCurrent pins the non-conflict path: +// a write carrying the current stored updated_at (or later) succeeds. +func TestUserStoreSetConditionalWriteAcceptsCurrent(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + firstUpdatedAt, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v1"`), nil) + if err != nil { + t.Fatalf("first set: %v", err) + } + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v2"`), &firstUpdatedAt); err != nil { + t.Fatalf("second set (current updatedAt): %v", err) + } + + got, _, _, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("get: %v", err) + } + if string(got) != `"v2"` { + t.Fatalf("got %q, want \"v2\"", got) + } +} + +// TestUserStoreUpdatedAtRoundTripsWithSubSecondPrecision pins that the +// updated_at Set returns is the same instant Get reads back. +// +// updated_at is persisted as a string, so the layout it is formatted with +// decides the stored resolution. time.RFC3339 carries no fractional seconds, +// which silently truncates every timestamp to a whole second: Set would +// return 06:21:17.825462712Z while Get returned 06:21:17Z for the same row. +// That mismatch is not cosmetic — ifUnmodifiedSince compares against the +// stored value, so a coarser resolution directly widens the window in which +// userStoreHasNewerRow cannot see a modification (see the same-second test +// below). +func TestUserStoreUpdatedAtRoundTripsWithSubSecondPrecision(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + written, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v1"`), nil) + if err != nil { + t.Fatalf("set: %v", err) + } + + _, read, found, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil || !found { + t.Fatalf("get: err=%v found=%v", err, found) + } + + if !read.Equal(written) { + t.Fatalf("updated_at lost precision in storage: Set returned %s, Get returned %s (delta %s)", + written.Format(time.RFC3339Nano), read.Format(time.RFC3339Nano), written.Sub(read)) + } +} + +// TestUserStateTimeLayoutPreservesLexicographicOrder pins the exact ordering +// bug found in review: time.RFC3339Nano trims trailing zeros from the +// fractional part, so a value landing on a "round" nanosecond (a trailing +// zero) formats shorter than one that doesn't — even when the round value is +// chronologically later. updated_at is compared as TEXT in the +// ifUnmodifiedSince WHERE clause, so that reordering would let a stale +// conditional write through. userStateTimeLayout's fixed width must not +// reproduce it. +func TestUserStateTimeLayoutPreservesLexicographicOrder(t *testing.T) { + earlier := time.Date(2026, 8, 1, 10, 0, 0, 123456780, time.UTC) // trailing zero trims under RFC3339Nano + later := time.Date(2026, 8, 1, 10, 0, 0, 123456781, time.UTC) // 1ns later, no trailing zero + + naiveEarlier, naiveLater := earlier.Format(time.RFC3339Nano), later.Format(time.RFC3339Nano) + if naiveLater >= naiveEarlier { + t.Fatalf( + "test setup invalid: expected time.RFC3339Nano to reproduce the ordering bug (later < earlier), got %q then %q", + naiveEarlier, naiveLater, + ) + } + + fixedEarlier, fixedLater := earlier.Format(userStateTimeLayout), later.Format(userStateTimeLayout) + if fixedEarlier >= fixedLater { + t.Fatalf("userStateTimeLayout does not preserve chronological order: %q then %q", fixedEarlier, fixedLater) + } +} + +// TestUserStoreSetConditionalWriteDetectsSameSecondModification is the +// regression test for the AC28 blind spot found in QA. +// +// TestUserStoreSetConditionalWriteConflict above passes only because it +// backdates ifUnmodifiedSince by a whole minute. The real scenario is much +// tighter: two of the same user's surfaces (task panel + kanban modal, or a +// debounced autosave firing every ~500ms) write within the same second. +// +// Sequence: tab A reads the note, tab B writes it, tab A then writes with the +// token it read. Tab B's write happened after tab A's read, so tab A must be +// told it lost the race (ErrConflict) rather than silently destroying tab B's +// edit. With second-resolution storage both writes share one timestamp, so +// `updatedAt.After(since)` is false and the guard never fires. +func TestUserStoreSetConditionalWriteDetectsSameSecondModification(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v0"`), nil); err != nil { + t.Fatalf("seed set: %v", err) + } + + // Tab A reads; this timestamp is the token it will later submit. + _, tabAToken, _, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("tab A get: %v", err) + } + + // Tab B writes immediately afterwards — same wall-clock second. + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"tabB-edit"`), nil); err != nil { + t.Fatalf("tab B set: %v", err) + } + + // Tab A writes with its now-stale token: must be rejected. + _, err = store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"tabA-edit"`), &tabAToken) + if !errors.Is(err, ErrConflict) { + t.Fatalf("tab A's stale conditional write was accepted (err=%v); tab B's edit is silently lost", err) + } + + stored, _, _, err := store.Get(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note") + if err != nil { + t.Fatalf("final get: %v", err) + } + if string(stored) != `"tabB-edit"` { + t.Fatalf("stored value = %s, want \"tabB-edit\" preserved", stored) + } +} + +// TestUserStoreSetConditionalWriteAllowsFirstWrite pins that +// ifUnmodifiedSince is a no-op when no row exists yet — a plugin creating a +// document for the first time never needs a nil-vs-zero special case. +func TestUserStoreSetConditionalWriteAllowsFirstWrite(t *testing.T) { + store := newTestUserStore(t) + ctx := context.Background() + + since := time.Now().UTC() + if _, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v1"`), &since); err != nil { + t.Fatalf("first set with ifUnmodifiedSince and no existing row: %v", err) + } +} + +func TestUserStoreInitSchemaIsIdempotent(t *testing.T) { + conn := newUserStoreSQLite(t) + if _, err := NewUserStore(db.NewPool(conn, conn)); err != nil { + t.Fatalf("first NewUserStore: %v", err) + } + if _, err := NewUserStore(db.NewPool(conn, conn)); err != nil { + t.Fatalf("second NewUserStore (re-init on existing schema): %v", err) + } +} + +// openIsolatedPostgresConcurrentPool is like testutil.OpenIsolatedPostgres +// but for tests that need real concurrent connections. OpenIsolatedPostgres +// isolates its schema with a session-level `SET search_path`, which only +// applies to the one connection it ran on; raising that *sqlx.DB's +// MaxOpenConns afterwards lets the pool open additional physical +// connections that never ran the SET and fall back to the default schema, +// so plugin_user_state "doesn't exist" the moment a query lands on a fresh +// connection (caught by actually running this test against a live +// container, not just trusting the retry-until-skip path). Baking +// search_path into the DSN via the `options` startup parameter instead +// applies it to every new physical connection at connect time. +func openIsolatedPostgresConcurrentPool(t *testing.T, dsn string, maxOpenConns int) *sqlx.DB { + t.Helper() + schema := "kandev_test_" + strings.ReplaceAll(uuid.NewString(), "-", "") + + setup, err := sqlx.Open("pgx", dsn) + if err != nil { + t.Fatalf("open postgres (schema setup): %v", err) + } + if _, err := setup.Exec("CREATE SCHEMA " + schema); err != nil { + _ = setup.Close() + t.Fatalf("create postgres schema %s: %v", schema, err) + } + t.Cleanup(func() { + _, _ = setup.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + _ = setup.Close() + }) + + pooled, err := sqlx.Open("pgx", dsn+" options='-c search_path="+schema+"'") + if err != nil { + t.Fatalf("open postgres (pooled): %v", err) + } + pooled.SetMaxOpenConns(maxOpenConns) + t.Cleanup(func() { _ = pooled.Close() }) + return pooled +} + +// TestUserStorePostgresConditionalWriteIsRaceFreeUnderConcurrency is the +// regression test for review feedback on Set: a plain read-then-write +// (even wrapped in one transaction) only serializes against Postgres's +// default READ COMMITTED isolation if the read takes a row lock. A bare +// SELECT does not, so two concurrent transactions can each see the row as +// not-yet-conflicting and both proceed to write, silently clobbering one +// another. Set instead folds the ifUnmodifiedSince comparison into the +// upsert's own WHERE clause, making the check-and-write one atomic, +// row-locking statement. Racing many goroutines against the same +// ifUnmodifiedSince token must let exactly one succeed. Skips unless +// KANDEV_TEST_POSTGRES_DSN is set. +func TestUserStorePostgresConditionalWriteIsRaceFreeUnderConcurrency(t *testing.T) { + conn := openIsolatedPostgresConcurrentPool(t, testutil.PostgresDSNFromEnv(t), 8) + store, err := NewUserStore(db.NewPool(conn, conn)) + if err != nil { + t.Fatalf("new user store: %v", err) + } + ctx := context.Background() + + seededAt, err := store.Set(ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", json.RawMessage(`"v0"`), nil) + if err != nil { + t.Fatalf("seed: %v", err) + } + + const attempts = 20 + var wg sync.WaitGroup + errs := make([]error, attempts) + for i := range attempts { + wg.Add(1) + go func(i int) { + defer wg.Done() + _, errs[i] = store.Set( + ctx, "kandev-plugin-notes", "user_1", "task", "task_xyz", "note", + json.RawMessage(fmt.Sprintf(`"racer-%d"`, i)), &seededAt, + ) + }(i) + } + wg.Wait() + + successes := 0 + for _, err := range errs { + switch { + case err == nil: + successes++ + case errors.Is(err, ErrConflict): + // expected for every loser of the race + default: + t.Fatalf("unexpected error from concurrent conditional write: %v", err) + } + } + if successes != 1 { + t.Fatalf("expected exactly 1 of %d concurrent conditional writes (same ifUnmodifiedSince) to succeed, got %d", + attempts, successes) + } +} + +func newTestUserStore(t *testing.T) *UserStore { + t.Helper() + conn := newUserStoreSQLite(t) + store, err := NewUserStore(db.NewPool(conn, conn)) + if err != nil { + t.Fatalf("new user store: %v", err) + } + return store +} + +func newUserStoreSQLite(t *testing.T) *sqlx.DB { + t.Helper() + conn, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} diff --git a/apps/backend/internal/plugins/user_state_handlers.go b/apps/backend/internal/plugins/user_state_handlers.go new file mode 100644 index 0000000000..64563e6933 --- /dev/null +++ b/apps/backend/internal/plugins/user_state_handlers.go @@ -0,0 +1,309 @@ +package plugins + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "regexp" + "time" + + "github.com/gin-gonic/gin" + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/auth/authn" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/plugins/state" + "github.com/kandev/kandev/internal/plugins/store" +) + +// maxUserStateBodyBytes caps the request body PUT +// /api/plugins/:id/user-state/... will read before persisting it, mirroring +// maxWebhookBodyBytes's rationale: without a cap, an authenticated-but- +// arbitrary browser write could exhaust backend memory. 256 KiB comfortably +// covers a rich-text scratchpad document (the driving use case, see +// docs/decisions/2026-08-01-per-user-plugin-storage.md) while bounding +// worst-case memory per request. +const maxUserStateBodyBytes = 256 << 10 // 256 KiB + +// userStateSegmentPattern validates the :scopeId and :key path segments of +// the per-user storage routes (AC18): non-empty, alphanumeric plus a small +// set of separators, capped at 128 characters. +var userStateSegmentPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`) + +// userStateScopes is the closed set of valid :scope values (AC18). +var userStateScopes = map[string]bool{ + "instance": true, + "workspace": true, + "task": true, + "session": true, + "repository": true, +} + +// registerUserStateRoutes wires the authenticated, per-user plugin storage +// surface (Approach D1: AC15-AC20, AC24, AC27, AC28). Registered before the +// public /:id/webhooks/:key wildcard, mirroring the /settings ordering +// caveat already documented in RegisterRoutes — these are static-ish +// sub-paths of /:id that must not be shadowed by a broader wildcard +// registered first. +func registerUserStateRoutes(api *gin.RouterGroup, ctrl *Controller) { + api.GET("/:id/user-state/:scope/:scopeId", ctrl.userStateList) + api.GET("/:id/user-state/:scope/:scopeId/:key", ctrl.userStateGet) + api.PUT("/:id/user-state/:scope/:scopeId/:key", ctrl.userStateSet) + api.DELETE("/:id/user-state/:scope/:scopeId/:key", ctrl.userStateDelete) +} + +// userStateSetRequest is the PUT .../user-state/:scope/:scopeId/:key body. +// Value is wrapped rather than being the bare request body so an arbitrary +// JSON value (array, string, number — not just an object) round-trips +// unambiguously alongside the sibling writerId/ifUnmodifiedSince fields. +type userStateSetRequest struct { + Value json.RawMessage `json:"value"` + WriterID string `json:"writerId,omitempty"` + IfUnmodifiedSince *time.Time `json:"ifUnmodifiedSince,omitempty"` +} + +// pluginUserStateUpdatedEvent is the bus event payload published on a +// successful PUT/DELETE (Approach F1, AC24). +// +// UserID is exported (json:"user_id") so it survives the JSON marshal a +// NATS-backed EventBus performs before publishing +// (internal/events/bus/nats.go): an unexported field disappears from that +// wire representation, so a NATS subscriber decodes event.Data as a map with +// no "user_id" at all, and UserEventBroadcaster silently drops the +// notification. It never reaches the browser regardless of transport — +// UserEventBroadcaster.subscribe strips "user_id" from the outgoing WS +// payload before building the message the client receives (routing is the +// broadcaster's concern; the client already knows its own identity). +type pluginUserStateUpdatedEvent struct { + UserID string `json:"user_id"` + PluginID string `json:"pluginId"` + Scope string `json:"scope"` + ScopeID string `json:"scopeId"` + Key string `json:"key"` + UpdatedAt time.Time `json:"updatedAt"` + WriterID string `json:"writerId,omitempty"` + Deleted bool `json:"deleted,omitempty"` +} + +// userStateIdentity resolves the caller's identity and the plugin record, +// enforces the record is active and declares the user_state capability, and +// validates :scope. It writes the appropriate error response and returns +// ok=false on any failure, so callers can return immediately. +func (c *Controller) userStateIdentity(ctx *gin.Context) (userID string, record *store.Record, scope string, ok bool) { + identity, hasIdentity := authn.FromGin(ctx) + if !hasIdentity { + ctx.JSON(http.StatusUnauthorized, gin.H{"error": "authentication required"}) + return "", nil, "", false + } + + rec, err := c.svc.Get(ctx.Param("id")) + if err != nil { + c.writeLookupError(ctx, err) + return "", nil, "", false + } + if rec.Status != StatusActive { + ctx.JSON(http.StatusNotFound, gin.H{"error": fmt.Sprintf("plugin %q is not active", rec.ID)}) + return "", nil, "", false + } + if !rec.Capabilities.UserState { + ctx.JSON(http.StatusForbidden, gin.H{ + "error": fmt.Sprintf("plugin %q does not declare the user_state capability", rec.ID), + }) + return "", nil, "", false + } + + scope = ctx.Param("scope") + if !userStateScopes[scope] { + ctx.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("invalid scope %q", scope)}) + return "", nil, "", false + } + + if c.svc.UserState() == nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": "user state store not configured"}) + return "", nil, "", false + } + + return identity.UserID, rec, scope, true +} + +// userStateSegments validates the :scopeId and (for single-key routes) :key +// path params against userStateSegmentPattern (AC18). +func (c *Controller) userStateSegments(ctx *gin.Context) (scopeID, key string, ok bool) { + scopeID = ctx.Param("scopeId") + if !userStateSegmentPattern.MatchString(scopeID) { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid scopeId"}) + return "", "", false + } + key = ctx.Param("key") + if !userStateSegmentPattern.MatchString(key) { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid key"}) + return "", "", false + } + return scopeID, key, true +} + +// userStateList serves GET /api/plugins/:id/user-state/:scope/:scopeId +// (AC27): every entry under that scope, ordered by key. +func (c *Controller) userStateList(ctx *gin.Context) { + userID, record, scope, ok := c.userStateIdentity(ctx) + if !ok { + return + } + scopeID := ctx.Param("scopeId") + if !userStateSegmentPattern.MatchString(scopeID) { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid scopeId"}) + return + } + + entries, err := c.svc.UserState().List(ctx.Request.Context(), record.ID, userID, scope, scopeID) + if err != nil { + c.log.Warn("plugin user-state list error", zap.Error(err)) + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + if entries == nil { + entries = []state.UserStateEntry{} + } + ctx.JSON(http.StatusOK, gin.H{"entries": entries}) +} + +// userStateGet serves GET /api/plugins/:id/user-state/:scope/:scopeId/:key +// (AC15's positive path). +func (c *Controller) userStateGet(ctx *gin.Context) { + userID, record, scope, ok := c.userStateIdentity(ctx) + if !ok { + return + } + scopeID, key, ok := c.userStateSegments(ctx) + if !ok { + return + } + + value, updatedAt, found, err := c.svc.UserState().Get(ctx.Request.Context(), record.ID, userID, scope, scopeID, key) + if err != nil { + c.log.Warn("plugin user-state get error", zap.Error(err)) + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + if !found { + ctx.JSON(http.StatusNotFound, gin.H{"error": "not found"}) + return + } + ctx.JSON(http.StatusOK, gin.H{"value": value, "updatedAt": updatedAt}) +} + +// userStateSet serves PUT /api/plugins/:id/user-state/:scope/:scopeId/:key +// (AC15's write path, AC28's conditional-write path, AC24's publish). +func (c *Controller) userStateSet(ctx *gin.Context) { + userID, record, scope, ok := c.userStateIdentity(ctx) + if !ok { + return + } + scopeID, key, ok := c.userStateSegments(ctx) + if !ok { + return + } + + body, err := readCappedUserStateBody(ctx) + if err != nil { + return // readCappedUserStateBody already wrote the error response + } + + var req userStateSetRequest + if err := json.Unmarshal(body, &req); err != nil || len(req.Value) == 0 { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload: value required"}) + return + } + + updatedAt, err := c.svc.UserState().Set( + ctx.Request.Context(), record.ID, userID, scope, scopeID, key, req.Value, req.IfUnmodifiedSince, + ) + if err != nil { + if errors.Is(err, state.ErrConflict) { + ctx.JSON(http.StatusConflict, gin.H{"error": "conflict: value was modified since ifUnmodifiedSince"}) + return + } + c.log.Warn("plugin user-state set error", zap.Error(err)) + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + c.publishUserStateUpdated(ctx, record.ID, userID, scope, scopeID, key, updatedAt, req.WriterID, false) + ctx.JSON(http.StatusOK, gin.H{"updatedAt": updatedAt}) +} + +// userStateDelete serves DELETE +// /api/plugins/:id/user-state/:scope/:scopeId/:key. writerId, if the caller +// wants echo suppression on delete too, is passed as a query param since +// DELETE carries no body in this API. +func (c *Controller) userStateDelete(ctx *gin.Context) { + userID, record, scope, ok := c.userStateIdentity(ctx) + if !ok { + return + } + scopeID, key, ok := c.userStateSegments(ctx) + if !ok { + return + } + + if err := c.svc.UserState().Delete(ctx.Request.Context(), record.ID, userID, scope, scopeID, key); err != nil { + c.log.Warn("plugin user-state delete error", zap.Error(err)) + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + + c.publishUserStateUpdated(ctx, record.ID, userID, scope, scopeID, key, time.Now().UTC(), ctx.Query("writerId"), true) + ctx.JSON(http.StatusOK, gin.H{"deleted": true}) +} + +// publishUserStateUpdated best-effort publishes the realtime notification +// (Approach F1) after a successful write/delete. A nil/unconfigured event +// bus is a silent no-op — the write itself already succeeded. +func (c *Controller) publishUserStateUpdated( + ctx *gin.Context, pluginID, userID, scope, scopeID, key string, updatedAt time.Time, writerID string, deleted bool, +) { + eventBus := c.svc.EventBus() + if eventBus == nil { + return + } + payload := pluginUserStateUpdatedEvent{ + UserID: userID, + PluginID: pluginID, + Scope: scope, + ScopeID: scopeID, + Key: key, + UpdatedAt: updatedAt, + WriterID: writerID, + Deleted: deleted, + } + evt := bus.NewEvent(events.PluginUserStateUpdated, "plugins-service", payload) + if err := eventBus.Publish(ctx.Request.Context(), events.PluginUserStateUpdated, evt); err != nil { + c.log.Warn("plugin user-state event publish failed", zap.Error(err)) + } +} + +// readCappedUserStateBody reads ctx.Request.Body bounded at +// maxUserStateBodyBytes via http.MaxBytesReader, writing the 413 response +// itself (and returning a non-nil error as a sentinel to the caller) when +// the body exceeds the cap — mirrors readCappedWebhookBody's approach for +// the same reason (unbounded io.ReadAll on an authenticated-but-arbitrary +// write is still a memory-exhaustion vector). +func readCappedUserStateBody(ctx *gin.Context) ([]byte, error) { + ctx.Request.Body = http.MaxBytesReader(ctx.Writer, ctx.Request.Body, maxUserStateBodyBytes) + body, err := io.ReadAll(ctx.Request.Body) + if err != nil { + var tooLarge *http.MaxBytesError + if errors.As(err, &tooLarge) { + ctx.JSON(http.StatusRequestEntityTooLarge, gin.H{ + "error": fmt.Sprintf("request body exceeds max size of %d bytes", maxUserStateBodyBytes), + }) + return nil, err + } + ctx.JSON(http.StatusBadRequest, gin.H{"error": "failed to read request body"}) + return nil, err + } + return body, nil +} diff --git a/apps/backend/internal/plugins/user_state_handlers_test.go b/apps/backend/internal/plugins/user_state_handlers_test.go new file mode 100644 index 0000000000..4ab798e29d --- /dev/null +++ b/apps/backend/internal/plugins/user_state_handlers_test.go @@ -0,0 +1,365 @@ +package plugins + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + goruntime "runtime" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + + "github.com/kandev/kandev/internal/auth/authn" + "github.com/kandev/kandev/internal/db" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/plugins/pkgtar/pkgtartest" + "github.com/kandev/kandev/internal/plugins/state" +) + +// newTestUserStateStore returns an in-memory-sqlite-backed *state.UserStore. +func newTestUserStateStore(t *testing.T) *state.UserStore { + t.Helper() + conn, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { _ = conn.Close() }) + + st, err := state.NewUserStore(db.NewPool(conn, conn)) + if err != nil { + t.Fatalf("new user state store: %v", err) + } + return st +} + +// userStatePluginPackage builds a valid, runtime-managed plugin tar.gz +// declaring (or not) the user_state capability. +func userStatePluginPackage(t *testing.T, id string, userStateEnabled bool) *bytes.Buffer { + t.Helper() + platformKey := goruntime.GOOS + "-" + goruntime.GOARCH + manifestYAML := fmt.Sprintf(` +id: %s +api_version: 1 +version: 1.0.0 +display_name: Test Plugin +capabilities: + user_state: %v +runtime: + type: binary + executables: + %s: server/plugin +`, id, userStateEnabled, platformKey) + + var buf bytes.Buffer + files := map[string][]byte{ + "manifest.yaml": []byte(manifestYAML), + "server/plugin": []byte("#!/bin/sh\necho fake\n"), + } + if err := pkgtartest.WritePackage(&buf, files); err != nil { + t.Fatalf("WritePackage: %v", err) + } + return &buf +} + +// newUserStateTestRouter wires a Service (with UserState + EventBus) behind +// a router that injects an identity from the X-Test-User-ID header +// (defaulting to "user_1"), mirroring how the real global auth middleware +// injects authn.Identity before plugin routes run. +func newUserStateTestRouter(t *testing.T) (*gin.Engine, *Service, bus.EventBus) { + t.Helper() + gin.SetMode(gin.TestMode) + svc, _, _ := newTestService(t) + svc.SetUserState(newTestUserStateStore(t)) + memBus := bus.NewMemoryEventBus(testLogger(t)) + svc.eventBus = memBus + + router := gin.New() + router.Use(func(c *gin.Context) { + userID := c.GetHeader("X-Test-User-ID") + if userID == "" { + userID = "user_1" + } + authn.SetOnGin(c, authn.Identity{UserID: userID, Role: authn.RoleMember}) + c.Next() + }) + RegisterRoutes(router, svc, nil, testLogger(t)) + return router, svc, memBus +} + +func installUserStatePlugin(t *testing.T, svc *Service, id string, userStateEnabled bool) { + t.Helper() + if _, err := svc.Install(context.Background(), userStatePluginPackage(t, id, userStateEnabled)); err != nil { + t.Fatalf("install %q: %v", id, err) + } +} + +func userStateReq(router *gin.Engine, method, path, userID, body string) *httptest.ResponseRecorder { + req := httptest.NewRequest(method, path, strings.NewReader(body)) + if userID != "" { + req.Header.Set("X-Test-User-ID", userID) + } + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec +} + +// TestUserStatePUTThenGETRoundTripsForCallingUser pins AC15's positive path. +func TestUserStatePUTThenGETRoundTripsForCallingUser(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", `{"value":"hello"}`) + if rec.Code != http.StatusOK { + t.Fatalf("PUT status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + + rec = userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", "") + if rec.Code != http.StatusOK { + t.Fatalf("GET status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + var resp struct { + Value json.RawMessage `json:"value"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if string(resp.Value) != `"hello"` { + t.Fatalf("value = %q, want %q", resp.Value, `"hello"`) + } +} + +// TestUserStateGETFromDifferentUserReturns404 pins AC15's cross-user negative. +func TestUserStateGETFromDifferentUserReturns404(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", `{"value":"alice's note"}`) + if rec.Code != http.StatusOK { + t.Fatalf("PUT status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + + rec = userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_2", "") + if rec.Code != http.StatusNotFound { + t.Fatalf("GET (different user) status = %d, want 404, body=%s", rec.Code, rec.Body.String()) + } +} + +// TestUserStateWithoutCapabilityReturns403 pins AC17. +func TestUserStateWithoutCapabilityReturns403(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-no-user-state", false) + + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-no-user-state/user-state/task/task_1/note", "user_1", `{"value":"x"}`) + if rec.Code != http.StatusForbidden { + t.Fatalf("status = %d, want 403, body=%s", rec.Code, rec.Body.String()) + } +} + +// TestUserStateUnknownPluginReturns404 and disabled-plugin variant pin AC17's +// 404 half. +func TestUserStateUnknownPluginReturns404(t *testing.T) { + router, _, _ := newUserStateTestRouter(t) + + rec := userStateReq(router, http.MethodGet, "/api/plugins/does-not-exist/user-state/task/task_1/note", "user_1", "") + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404, body=%s", rec.Code, rec.Body.String()) + } +} + +func TestUserStateDisabledPluginReturns404(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + if err := svc.Disable("kandev-plugin-notes"); err != nil { + t.Fatalf("disable: %v", err) + } + + rec := userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", "") + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404, body=%s", rec.Code, rec.Body.String()) + } +} + +// TestUserStateInvalidScopeReturns400 and TestUserStateInvalidScopeIDReturns400 +// pin AC18. +func TestUserStateInvalidScopeReturns400(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + rec := userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/bogus-scope/task_1/note", "user_1", "") + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400, body=%s", rec.Code, rec.Body.String()) + } +} + +func TestUserStateInvalidScopeIDReturns400(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + // A leading "." fails the [A-Za-z0-9] leading-char requirement. + rec := userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/task/.bad/note", "user_1", "") + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400, body=%s", rec.Code, rec.Body.String()) + } +} + +func TestUserStateInvalidKeyReturns400(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/.bad-key", "user_1", `{"value":"x"}`) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400, body=%s", rec.Code, rec.Body.String()) + } +} + +// TestUserStateOversizedBodyReturns413 pins AC18's body cap. +func TestUserStateOversizedBodyReturns413(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + huge := `{"value":"` + strings.Repeat("x", maxUserStateBodyBytes+1) + `"}` + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", huge) + if rec.Code != http.StatusRequestEntityTooLarge { + t.Fatalf("status = %d, want 413, body=%s", rec.Code, rec.Body.String()) + } +} + +// TestUserStateListReturnsOrderedEntries pins AC27. +func TestUserStateListReturnsOrderedEntries(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + for _, key := range []string{"zeta", "alpha"} { + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/"+key, "user_1", `{"value":1}`) + if rec.Code != http.StatusOK { + t.Fatalf("PUT %s status = %d, body=%s", key, rec.Code, rec.Body.String()) + } + } + + rec := userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/task/task_1", "user_1", "") + if rec.Code != http.StatusOK { + t.Fatalf("LIST status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + var resp struct { + Entries []state.UserStateEntry `json:"entries"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(resp.Entries) != 2 || resp.Entries[0].Key != "alpha" || resp.Entries[1].Key != "zeta" { + t.Fatalf("entries = %+v, want [alpha, zeta] order", resp.Entries) + } +} + +// TestUserStateConditionalWriteConflictReturns409 pins AC28. +func TestUserStateConditionalWriteConflictReturns409(t *testing.T) { + router, svc, _ := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", `{"value":"v1"}`) + if rec.Code != http.StatusOK { + t.Fatalf("first PUT status = %d, body=%s", rec.Code, rec.Body.String()) + } + + stale := `{"value":"v2","ifUnmodifiedSince":"2000-01-01T00:00:00Z"}` + rec = userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", stale) + if rec.Code != http.StatusConflict { + t.Fatalf("conflicting PUT status = %d, want 409, body=%s", rec.Code, rec.Body.String()) + } + + rec = userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", "") + var resp struct { + Value json.RawMessage `json:"value"` + } + _ = json.Unmarshal(rec.Body.Bytes(), &resp) + if string(resp.Value) != `"v1"` { + t.Fatalf("value after conflicting write = %q, want unchanged \"v1\"", resp.Value) + } +} + +// TestUserStatePUTPublishesEventScopedToWriter pins AC24: a successful PUT +// publishes plugin.user-state.updated carrying the writer's user_id. +// +// user_id is deliberately readable here, on the raw published bus event — +// that is what lets UserEventBroadcaster route it under a NATS-backed +// EventBus, which marshals every event before publishing and would silently +// drop an unexported field (see pluginUserStateUpdatedEvent's doc comment). +// The "never reaches the browser" half of that contract is enforced one +// layer up, in the broadcaster itself: see +// internal/gateway/websocket/user_notifications_test.go's +// TestRegisterUserNotifications_PluginUserStatePayloadShape and its +// NATS-transport-simulating sibling. +func TestUserStatePUTPublishesEventScopedToWriter(t *testing.T) { + router, svc, memBus := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + var gotPayload map[string]any + _, err := memBus.Subscribe(events.PluginUserStateUpdated, func(_ context.Context, evt *bus.Event) error { + raw, _ := json.Marshal(evt.Data) + _ = json.Unmarshal(raw, &gotPayload) + return nil + }) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", `{"value":"hi","writerId":"tab-42"}`) + if rec.Code != http.StatusOK { + t.Fatalf("PUT status = %d, body=%s", rec.Code, rec.Body.String()) + } + + if gotPayload["user_id"] != "user_1" { + t.Fatalf("published event user_id = %v, want %q (must survive marshal for NATS routing)", gotPayload["user_id"], "user_1") + } + if gotPayload["pluginId"] != "kandev-plugin-notes" || gotPayload["scope"] != "task" || + gotPayload["scopeId"] != "task_1" || gotPayload["key"] != "note" || gotPayload["writerId"] != "tab-42" { + t.Fatalf("event payload = %+v, missing expected fields", gotPayload) + } + if deleted, _ := gotPayload["deleted"].(bool); deleted { + t.Fatalf("PUT event payload.deleted = true, want false/absent") + } +} + +// TestUserStateDELETEPublishesDeletedEvent covers the DELETE half of AC24. +func TestUserStateDELETEPublishesDeletedEvent(t *testing.T) { + router, svc, memBus := newUserStateTestRouter(t) + installUserStatePlugin(t, svc, "kandev-plugin-notes", true) + + var gotDeleted bool + _, err := memBus.Subscribe(events.PluginUserStateUpdated, func(_ context.Context, evt *bus.Event) error { + raw, _ := json.Marshal(evt.Data) + var payload map[string]any + _ = json.Unmarshal(raw, &payload) + gotDeleted, _ = payload["deleted"].(bool) + return nil + }) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + + rec := userStateReq(router, http.MethodPut, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", `{"value":"hi"}`) + if rec.Code != http.StatusOK { + t.Fatalf("PUT status = %d, body=%s", rec.Code, rec.Body.String()) + } + rec = userStateReq(router, http.MethodDelete, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", "") + if rec.Code != http.StatusOK { + t.Fatalf("DELETE status = %d, want 200, body=%s", rec.Code, rec.Body.String()) + } + if !gotDeleted { + t.Fatalf("expected DELETE to publish an event with deleted=true") + } + + rec = userStateReq(router, http.MethodGet, "/api/plugins/kandev-plugin-notes/user-state/task/task_1/note", "user_1", "") + if rec.Code != http.StatusNotFound { + t.Fatalf("GET after delete status = %d, want 404", rec.Code) + } +} diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 16bd3d52ba..82fd44da89 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -325,6 +325,12 @@ const ( ActionUserSettingsUpdate = "user.settings.update" ActionUserSettingsUpdated = "user.settings.updated" + // ActionPluginUserStateUpdated notifies the writing user's other WS + // connections that one of their per-user plugin storage keys changed + // (Approach F1). Payload: {pluginId, scope, scopeId, key, updatedAt, + // writerId, deleted} — keys only, never the stored value. + ActionPluginUserStateUpdated = "plugin.user-state.updated" + // System maintenance jobs (VACUUM, factory reset, snapshot create/restore, // disk walk). Broadcast to all connected clients so the System pages can // render progress. diff --git a/apps/web/AGENTS.md b/apps/web/AGENTS.md index 249a1284de..a7204c95f9 100644 --- a/apps/web/AGENTS.md +++ b/apps/web/AGENTS.md @@ -74,8 +74,7 @@ lib/api/domains/ # API clients **Hydration:** Go injects `window.__KANDEV_BOOT_PAYLOAD__` into the SPA shell before React mounts. `lib/state/hydration/merge-strategies.ts` has `deepMerge()`, `mergeSessionMap()`, `mergeLoadingState()` to avoid overwriting live client state. Pass `activeSessionId` to protect active sessions. -For rebasing or finishing PRs written against the old Next.js runtime, follow -[`docs/nextjs-spa-migration.md`](../../docs/nextjs-spa-migration.md). +For rebasing or finishing PRs written against the old Next.js runtime, follow [`docs/nextjs-spa-migration.md`](../../docs/nextjs-spa-migration.md). **Hooks Pattern:** Hooks in `hooks/domains/` encapsulate WS subscription + store selection. WS client deduplicates subscriptions automatically. @@ -150,21 +149,17 @@ surface. with independent open state. Touch-pinned help must close on a second trigger tap, outside interaction, and Escape; verify desktop pointer and mobile-sized touch flows. -- **Renaming a `data-testid`:** set the new id as `data-testid=""` and keep - the old id as `data-legacy-testid=""`, then migrate e2e specs to the new - id in the same PR. JSX rejects two `data-testid` attributes on one element, - and Playwright's `getByTestId` only matches one attribute name — the - `data-legacy-testid` alias lets existing specs keep selecting the element - while the migration is in flight. -- **Dockview session panel activation:** session chat panels can become active - through tab pointer/keyboard events, global tab-cycling shortcuts, - reopen/menu actions, and Dockview close controls. When changing - `tasks.activeSessionId` or active-session sync, audit all of those paths. Use - store state in addition to Dockview `api.isActive`; the current session's chat - tab may be Dockview-inactive while Files/Changes is active. Same-session - clicks must not leave stale activation intent, and Dockview - `.dv-default-tab-action` close controls should be treated as close/delete - actions rather than session-switch intent. +- **Renaming a `data-testid`:** set the new id as `data-testid=""` and keep the old id as + `data-legacy-testid=""`, then migrate e2e specs to the new id in the same PR. JSX rejects two + `data-testid` attributes on one element, and Playwright's `getByTestId` only matches one attribute + name — the `data-legacy-testid` alias lets existing specs keep selecting the element mid-migration. +- **Dockview session panel activation:** session chat panels can become active through tab + pointer/keyboard events, global tab-cycling shortcuts, reopen/menu actions, and Dockview close + controls. When changing `tasks.activeSessionId` or active-session sync, audit all of those paths. + Use store state in addition to Dockview `api.isActive`; the current session's chat tab may be + Dockview-inactive while Files/Changes is active. Same-session clicks must not leave stale + activation intent, and Dockview `.dv-default-tab-action` close controls should be treated as + close/delete actions rather than session-switch intent. - **Conditional review-panel ownership:** the reusable `pr-detail` panel may be visible only while the active task has a linked PR or MR. A custom Default layout's canonical panel supplies the preferred group and tab index; it does @@ -173,16 +168,14 @@ surface. Restoration, maximized layouts, and a session's offered/dismissed marker defer or suppress automatic insertion; linked existing panels synchronize provider and review identity without moving them. -- **GitHub PR status UI:** visual PR/CI status surfaces should use the shared - helpers in `apps/web/components/github/pr-task-icon.tsx` - (`hasPRChecksPassedForDisplay`, `hasPRChecksInProgressForDisplay`, and - `hasPRChecksPassedWithoutReviewWaitForDisplay`) instead of re-deriving status - from `checks_state`, `checks_total`, or `checks_passing` locally. Aggregate - check counts are a display-only fallback when `checks_state` is empty; they may - make chips or task icons render passed/in-progress, but must not enable merge - actions. Merge readiness must use `isPRReadyToMerge`, which requires GitHub's - explicit `checks_state === "success"` rollup. When changing PR status behavior, - update both `pr-task-icon.test.ts` and `pr-status-chip.test.tsx`. +- **GitHub PR status UI:** visual PR/CI status surfaces should use the shared helpers in + `apps/web/components/github/pr-task-icon.tsx` (`hasPRChecksPassedForDisplay`, + `hasPRChecksInProgressForDisplay`, `hasPRChecksPassedWithoutReviewWaitForDisplay`) instead of + re-deriving status from `checks_state`, `checks_total`, or `checks_passing` locally. Aggregate + check counts are a display-only fallback when `checks_state` is empty; they may make chips or task + icons render passed/in-progress, but must not enable merge actions. Merge readiness must use + `isPRReadyToMerge`, which requires GitHub's explicit `checks_state === "success"` rollup. When + changing PR status behavior, update both `pr-task-icon.test.ts` and `pr-status-chip.test.tsx`. - **Task repository labels:** user-facing task/card repo chips should display a stable repo slug or name (`owner/repo` when known, otherwise the repo name), not a local filesystem path. Local clone paths or folder paths belong in @@ -270,13 +263,38 @@ Enforced by `apps/web/eslint.config.mjs` (warnings, will become errors): When you hit a limit, extract a helper function, custom hook, or sub-component. Prefer composition over growing a single function. +## Plugin system + +The frozen frontend contract is `docs/plans/plugins/PLUGIN-API.md`; `lib/plugins/types.ts` +is its TS mirror — the two must change together. `lib/plugins/registry.ts` is the +reactive singleton `PluginRegistry`; every `register*` call there needs a matching +cleanup entry in `unregisterPlugin` and `totalCount()`, or a disabled/uninstalled +plugin leaks a stale registration. + +- **Task panels** (`registerTaskPanel`): one generic dockview component, `"plugin-panel"`, shared by + every plugin — identity lives in `params: { pluginId, panelKey }` (id helpers in + `lib/state/layout-manager/plugin-panels.ts`). `renderPanel` in `dockview-shared.tsx` and + `dockview-panel-content.tsx` each get exactly one `"plugin-panel"` case (lookup tables, not + switches). `PluginTaskPanel` (`components/task/`) resolves the registration behind a + `PluginErrorBoundary`; `mobileEnabled: true` also renders it via the phone bottom nav + (`session-mobile-bottom-nav.tsx`) with `presentation: "mobile"`. +- **Kanban card contributions:** `registerTaskMenuAction({ group: "edit", ... })` turns the flat + `Edit` item into an `Edit` submenu (`kanban-card-edit-submenu.tsx`); + `registerComponent("task-card-indicators", ...)` renders beside `PRTaskIcon` via ``. +- **`host.storage`:** authenticated per-user key/value storage (`lib/plugins/host-api.ts`), backed by + `/api/plugins/{id}/user-state/...` (`docs/decisions/2026-08-01-per-user-plugin-storage.md`). + `subscribe` (`lib/plugins/user-state-sync.ts`) wraps `registerWsHandler` with own-plugin filtering + and own-tab echo suppression via a per-tab `writerId`. +- **`host.ui.RichTextEditor`/`RichTextReadOnly`** (`components/editors/tiptap/rich-text-editor.tsx`): + narrow Plan-panel-tiptap wrappers — don't widen their props beyond + `{ taskId, value, onChange, placeholder, className, testId }` / `{ value, className, testId }` + without updating `PLUGIN-API.md`. + ## Testing notes - jsdom drops `secure` cookies over `http`, so `document.cookie` reads back empty. To assert a cookie write in a Vitest unit test, intercept the setter with `Object.defineProperty(document, "cookie", { set: ... })` and restore it after. -- jsdom synthetic mouse events do not reliably open Radix Tooltip. In component - tests, render under `TooltipProvider` and assert the keyboard-focus path with - `fireEvent.focus`. Cover pointer hover in Playwright with `locator.hover()` and - assert the visible portaled `role="tooltip"`; do not remove a hover regression - solely because `mouseenter` or `pointerMove` failed in jsdom. -- In Playwright tests, avoid strict locators that assume only one `terminal-panel` or `.xterm` exists. Mobile and dockview layouts can mount multiple terminal instances; scope to the active panel or use `.first()` / `.last()` deliberately with a comment or helper. -- Shared E2E helpers that inspect mounted React/DOM internals must be scoped to the active panel/container, not global selectors, because hidden or stale mounted panels can coexist in dock/mobile layouts. +- jsdom synthetic mouse events do not reliably open Radix Tooltip. In component tests, render under + `TooltipProvider` and assert the keyboard-focus path with `fireEvent.focus`. Cover pointer hover in + Playwright with `locator.hover()` and assert the portaled `role="tooltip"`; don't drop a hover + regression just because `mouseenter`/`pointerMove` failed in jsdom. +- In Playwright tests, avoid strict locators that assume only one `terminal-panel` or `.xterm` exists. Mobile and dockview layouts can mount multiple terminal instances; scope to the active panel or use `.first()` / `.last()` deliberately with a comment or helper. Shared E2E helpers that inspect mounted React/DOM internals must also be scoped to the active panel/container, not global selectors, because hidden or stale mounted panels can coexist in dock/mobile layouts. diff --git a/apps/web/components/editors/tiptap/rich-text-editor.test.tsx b/apps/web/components/editors/tiptap/rich-text-editor.test.tsx new file mode 100644 index 0000000000..c711bbfaff --- /dev/null +++ b/apps/web/components/editors/tiptap/rich-text-editor.test.tsx @@ -0,0 +1,73 @@ +import { cleanup, fireEvent, render, screen } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +vi.mock("./tiptap-plan-editor", () => ({ + TipTapPlanEditor: (props: { + taskId: string; + value: string; + onChange: (value: string) => void; + placeholder?: string; + }) => ( +