From 477b37211083f5a3a645c0ccd8fd43b9243c36ae Mon Sep 17 00:00:00 2001 From: mirivlad Date: Sat, 11 Jul 2026 23:30:38 +0800 Subject: [PATCH] fix: persist browser captures atomically --- .../src/lib/plugin-host/VerstakPluginAPI.js | 6 +- internal/api/app.go | 267 +++++++++++++++--- internal/api/app_test.go | 246 ++++++++++++++++ internal/core/browserreceiver/receiver.go | 43 ++- .../core/browserreceiver/receiver_test.go | 56 ++++ internal/core/storage/api.go | 61 +++- internal/core/vault/vault.go | 17 ++ 7 files changed, 641 insertions(+), 55 deletions(-) diff --git a/frontend/src/lib/plugin-host/VerstakPluginAPI.js b/frontend/src/lib/plugin-host/VerstakPluginAPI.js index eb4ab47..d5f56f7 100644 --- a/frontend/src/lib/plugin-host/VerstakPluginAPI.js +++ b/frontend/src/lib/plugin-host/VerstakPluginAPI.js @@ -236,12 +236,10 @@ export function createPluginAPI(pluginId) { if (!key) { throw new Error('settings.write requires a key'); } - const settings = await this.read(); - settings[key] = value; await callBackendErrorString(pluginId, 'settings.write(' + key + ')', function() { - return App.WritePluginSettings(pluginId, settings); + return App.WritePluginSetting(pluginId, key, value); }); - return settings; + return this.read(); }, writeAll: function(settings) { assertActive('settings.writeAll'); diff --git a/internal/api/app.go b/internal/api/app.go index b009bbf..8dcb4bd 100644 --- a/internal/api/app.go +++ b/internal/api/app.go @@ -9,7 +9,9 @@ import ( "net/url" "os" "path/filepath" + "sort" "strings" + "sync/atomic" "time" "github.com/wailsapp/wails/v2/pkg/runtime" @@ -41,6 +43,12 @@ const pluginEventRuntimeName = "verstak:plugin-event" const activityGlobalKey = "events:global" const activityWorkspacePrefix = "events:workspace:" const maxActivityEvents = 250 +const browserInboxPluginID = "verstak.browser-inbox" +const browserInboxGlobalKey = "captures:global" +const browserInboxLegacyKey = "captures" +const browserInboxWorkspacePrefix = "captures:workspace:" +const browserInboxMutationEvent = "browser-inbox.storage.mutate" +const maxBrowserInboxCaptures = 100 const workspaceCreatedEventName = "workspace.created" const workspaceRenamedEventName = "workspace.renamed" const workspaceTrashedEventName = "workspace.trashed" @@ -48,26 +56,28 @@ const workspaceSelectedEventName = "workspace.selected" // App is the main application struct exposed to the Wails frontend. type App struct { - ctx context.Context - capRegistry *capability.Registry - contribRegistry *contribution.Registry - permRegistry *permissions.Registry - eventBus *events.Bus - plugins []plugin.Plugin - vault *vault.Vault - storage *storage.Storage - files *corefiles.Service - externalOpen externalOpenService - appSettings *appsettings.Manager - pluginState *pluginstate.Manager - workbench *coreworkbench.Router - workspace *workspace.Manager - syncSvc *syncsvc.Service - browserReceiver *browserreceiver.Receiver - secretsSession *coresecrets.VaultSession - fileWatcher *filewatcher.Service - debug bool - activityEvents map[string]bool + ctx context.Context + capRegistry *capability.Registry + contribRegistry *contribution.Registry + permRegistry *permissions.Registry + eventBus *events.Bus + plugins []plugin.Plugin + vault *vault.Vault + storage *storage.Storage + files *corefiles.Service + externalOpen externalOpenService + appSettings *appsettings.Manager + pluginState *pluginstate.Manager + workbench *coreworkbench.Router + workspace *workspace.Manager + syncSvc *syncsvc.Service + browserReceiver *browserreceiver.Receiver + secretsSession *coresecrets.VaultSession + fileWatcher *filewatcher.Service + debug bool + activityEvents map[string]bool + browserInboxEvents map[string]bool + browserInboxEnabled atomic.Bool } type externalOpenService interface { @@ -93,29 +103,34 @@ func NewApp( debugEnabled bool, ) *App { app := &App{ - capRegistry: capReg, - contribRegistry: contribReg, - permRegistry: permReg, - eventBus: bus, - plugins: plugins, - vault: vaultService, - storage: storageService, - files: filesService, - externalOpen: externalopen.NewService(), - appSettings: appSettingsMgr, - pluginState: pluginStateMgr, - workbench: coreworkbench.NewRouter(workbenchPrefsFromSettings(appSettingsMgr)), - workspace: workspaceMgr, - syncSvc: syncService, - browserReceiver: browserReceiverService, - fileWatcher: filewatcher.NewService(bus, 0), - debug: debugEnabled, - activityEvents: make(map[string]bool), + capRegistry: capReg, + contribRegistry: contribReg, + permRegistry: permReg, + eventBus: bus, + plugins: plugins, + vault: vaultService, + storage: storageService, + files: filesService, + externalOpen: externalopen.NewService(), + appSettings: appSettingsMgr, + pluginState: pluginStateMgr, + workbench: coreworkbench.NewRouter(workbenchPrefsFromSettings(appSettingsMgr)), + workspace: workspaceMgr, + syncSvc: syncService, + browserReceiver: browserReceiverService, + fileWatcher: filewatcher.NewService(bus, 0), + debug: debugEnabled, + activityEvents: make(map[string]bool), + browserInboxEvents: make(map[string]bool), } if app.syncSvc == nil { app.rebindSyncService() } app.ensureActivityProviderSubscriptions() + app.ensureBrowserInboxSubscriptions() + if app.browserReceiver != nil { + app.browserReceiver.SetPersistence(app.browserInboxAvailable, app.recordBrowserCapture) + } app.startFileWatcherForOpenVault() return app } @@ -151,9 +166,184 @@ func (a *App) ensureWorkbench() *coreworkbench.Router { func (a *App) Startup(ctx context.Context) { a.ctx = ctx a.ensureActivityProviderSubscriptions() + a.ensureBrowserInboxSubscriptions() log.Printf("[api] App.Startup: initialized with %d plugins", len(a.plugins)) } +func (a *App) ensureBrowserInboxSubscriptions() { + if a.eventBus == nil || a.storage == nil { + a.browserInboxEnabled.Store(false) + return + } + if _, err := a.requirePluginAccess(browserInboxPluginID, "storage.namespace"); err != nil { + a.browserInboxEnabled.Store(false) + return + } + a.browserInboxEnabled.Store(true) + if a.browserInboxEvents == nil { + a.browserInboxEvents = make(map[string]bool) + } + for _, eventName := range []string{browserInboxMutationEvent} { + if a.browserInboxEvents[eventName] { + continue + } + a.browserInboxEvents[eventName] = true + a.eventBus.Subscribe(eventName, func(event events.Event) { + if err := a.mutateBrowserInboxCapture(event); err != nil { + log.Printf("[api] browser inbox mutation failed: %v", err) + } + }) + } +} + +func (a *App) browserInboxAvailable() bool { + if a == nil || a.storage == nil || a.vault == nil || !a.browserInboxEnabled.Load() || a.vault.GetVaultStatus() != vault.StatusOpen { + return false + } + return true +} + +func (a *App) recordBrowserCapture(event events.Event) error { + if !a.browserInboxAvailable() { + return fmt.Errorf("browser inbox unavailable") + } + capture := eventPayloadMap(event.Payload) + captureID := firstPayloadText(capture, "captureId") + if captureID == "" { + return fmt.Errorf("captureId is empty") + } + if firstPayloadText(capture, "kind") == "" { + capture["kind"] = strings.TrimPrefix(event.Name, "browser.capture.") + } + if firstPayloadText(capture, "capturedAt") == "" { + capture["capturedAt"] = event.Timestamp + } + capture["receivedAt"] = time.Now().UTC().Format(time.RFC3339Nano) + return a.updateBrowserInboxCaptures(func(captures []map[string]interface{}) []map[string]interface{} { + for _, stored := range captures { + if firstPayloadText(stored, "captureId") == captureID { + return captures + } + } + result := []map[string]interface{}{capture} + result = append(result, captures...) + return result + }) +} + +func (a *App) mutateBrowserInboxCapture(event events.Event) error { + payload := eventPayloadMap(event.Payload) + if firstPayloadText(payload, "pluginId") != browserInboxPluginID { + return fmt.Errorf("browser inbox mutation source is not authorized") + } + action := firstPayloadText(payload, "action") + switch action { + case "migrate", "assign", "delete", "processed": + default: + return fmt.Errorf("unsupported browser inbox mutation %q", action) + } + captureID := firstPayloadText(payload, "captureId") + captureIDs := make(map[string]bool) + if captureID != "" { + captureIDs[captureID] = true + } + if items, ok := payload["captureIds"].([]interface{}); ok { + for _, item := range items { + if id, ok := item.(string); ok && strings.TrimSpace(id) != "" { + captureIDs[strings.TrimSpace(id)] = true + } + } + } + if action != "migrate" && len(captureIDs) == 0 { + return fmt.Errorf("captureId is empty") + } + return a.updateBrowserInboxCaptures(func(captures []map[string]interface{}) []map[string]interface{} { + result := make([]map[string]interface{}, 0, len(captures)) + for _, capture := range captures { + storedID := firstPayloadText(capture, "captureId") + if !captureIDs[storedID] { + result = append(result, capture) + continue + } + switch action { + case "delete": + continue + case "assign": + workspaceRoot := firstPayloadText(payload, "workspaceRootPath") + capture["workspaceRootPath"] = workspaceRoot + capture["workspaceName"] = workspaceRoot + case "processed": + capture["processed"], _ = payload["processed"].(bool) + } + result = append(result, capture) + } + return result + }) +} + +func (a *App) updateBrowserInboxCaptures(update func([]map[string]interface{}) []map[string]interface{}) error { + if !a.browserInboxAvailable() { + return fmt.Errorf("browser inbox unavailable") + } + return a.storage.UpdatePluginSettings(browserInboxPluginID, func(settings map[string]interface{}) error { + captures, legacyKeys := browserInboxCaptures(settings) + captures = update(captures) + if len(captures) > maxBrowserInboxCaptures { + captures = captures[:maxBrowserInboxCaptures] + } + stored := make([]interface{}, 0, len(captures)) + for _, capture := range captures { + stored = append(stored, capture) + } + settings[browserInboxGlobalKey] = stored + for _, key := range legacyKeys { + settings[key] = []interface{}{} + } + return nil + }) +} + +func browserInboxCaptures(settings map[string]interface{}) ([]map[string]interface{}, []string) { + keys := []string{browserInboxGlobalKey, browserInboxLegacyKey} + for key := range settings { + if strings.HasPrefix(key, browserInboxWorkspacePrefix) { + keys = append(keys, key) + } + } + sort.Strings(keys[2:]) + seen := make(map[string]bool) + legacyKeys := make([]string, 0, len(keys)-1) + var captures []map[string]interface{} + for _, key := range keys { + if key != browserInboxGlobalKey { + legacyKeys = append(legacyKeys, key) + } + workspaceRoot := "" + if strings.HasPrefix(key, browserInboxWorkspacePrefix) { + workspaceRoot, _ = url.PathUnescape(strings.TrimPrefix(key, browserInboxWorkspacePrefix)) + } + items, _ := settings[key].([]interface{}) + for _, item := range items { + original, ok := item.(map[string]interface{}) + if !ok { + continue + } + capture := eventPayloadMap(original) + captureID := firstPayloadText(capture, "captureId") + if captureID == "" || seen[captureID] { + continue + } + seen[captureID] = true + if firstPayloadText(capture, "workspaceRootPath") == "" && workspaceRoot != "" { + capture["workspaceRootPath"] = workspaceRoot + capture["workspaceName"] = workspaceRoot + } + captures = append(captures, capture) + } + } + return captures, legacyKeys +} + func (a *App) findPlugin(pluginID string) (*plugin.Plugin, error) { for i := range a.plugins { if a.plugins[i].Manifest.ID == pluginID { @@ -646,6 +836,7 @@ func (a *App) ReloadPlugins() (int, string) { a.plugins = plugins a.ensureActivityProviderSubscriptions() + a.ensureBrowserInboxSubscriptions() var buf strings.Builder buf.WriteString("discovery complete") diff --git a/internal/api/app_test.go b/internal/api/app_test.go index 8fcf860..a5aa813 100644 --- a/internal/api/app_test.go +++ b/internal/api/app_test.go @@ -3,12 +3,14 @@ package api import ( "context" "encoding/json" + "fmt" "net" "net/http" "net/http/httptest" "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -617,6 +619,250 @@ func TestActivityProviderRecordsFileChangedWithoutMountedView(t *testing.T) { } } +func TestBrowserInboxRecordsCaptureWithoutMountedView(t *testing.T) { + v := vault.NewVault(nil) + if err := v.CreateVault(t.TempDir()); err != nil { + t.Fatalf("CreateVault: %v", err) + } + bus := events.NewBus() + app := &App{ + eventBus: bus, + storage: storage.New(v), + vault: v, + plugins: []plugin.Plugin{{ + Manifest: plugin.Manifest{ + ID: "verstak.browser-inbox", + Name: "Browser Inbox", + Permissions: []string{"storage.namespace"}, + }, + Status: plugin.StatusLoaded, + Enabled: true, + }}, + } + app.ensureBrowserInboxSubscriptions() + + if err := app.recordBrowserCapture(events.Event{ + Name: "browser.capture.page", + Timestamp: "2026-07-11T12:00:00Z", + Payload: map[string]interface{}{ + "captureId": "capture-background", + "capturedAt": "2026-07-11T11:59:00Z", + "kind": "page", + "url": "https://example.com/article", + "title": "Background capture", + "domain": "example.com", + }, + }); err != nil { + t.Fatalf("RecordBrowserCapture: %v", err) + } + bus.Publish(events.Event{Name: browserInboxMutationEvent, Payload: map[string]interface{}{ + "pluginId": browserInboxPluginID, "action": "assign", "captureId": "capture-background", "workspaceRootPath": "Project", + }}) + bus.Publish(events.Event{Name: browserInboxMutationEvent, Payload: map[string]interface{}{ + "pluginId": browserInboxPluginID, "action": "processed", "captureId": "capture-background", "processed": true, + }}) + if err := app.recordBrowserCapture(events.Event{ + Name: "browser.capture.page", + Payload: map[string]interface{}{ + "captureId": "capture-background", + "title": "Retried payload", + }, + }); err != nil { + t.Fatalf("RecordBrowserCapture retry: %v", err) + } + + settings, err := app.storage.ReadPluginSettings("verstak.browser-inbox") + if err != nil { + t.Fatalf("ReadPluginSettings: %v", err) + } + stored, ok := settings["captures:global"].([]interface{}) + if !ok || len(stored) != 1 { + t.Fatalf("captures:global = %#v, want one capture", settings["captures:global"]) + } + capture, ok := stored[0].(map[string]interface{}) + if !ok { + t.Fatalf("capture = %#v, want map[string]interface{}", stored[0]) + } + if capture["captureId"] != "capture-background" || capture["title"] != "Background capture" || capture["workspaceRootPath"] != "Project" || capture["processed"] != true { + t.Fatalf("stored capture = %#v", capture) + } +} + +func TestBrowserInboxRejectsCaptureWithoutOpenVault(t *testing.T) { + v := vault.NewVault(nil) + app := &App{ + storage: storage.New(v), + vault: v, + plugins: []plugin.Plugin{{ + Manifest: plugin.Manifest{ + ID: browserInboxPluginID, + Permissions: []string{"storage.namespace"}, + }, + Status: plugin.StatusLoaded, + Enabled: true, + }}, + } + app.ensureBrowserInboxSubscriptions() + + err := app.recordBrowserCapture(events.Event{ + Name: "browser.capture.page", + Payload: map[string]interface{}{"captureId": "capture-no-vault"}, + }) + + if err == nil || !strings.Contains(err.Error(), "unavailable") { + t.Fatalf("RecordBrowserCapture error = %v, want unavailable", err) + } +} + +func TestBrowserInboxSerializesConcurrentCapturesAndMutations(t *testing.T) { + v := vault.NewVault(nil) + if err := v.CreateVault(t.TempDir()); err != nil { + t.Fatalf("CreateVault: %v", err) + } + bus := events.NewBus() + app := &App{ + eventBus: bus, + storage: storage.New(v), + vault: v, + plugins: []plugin.Plugin{{ + Manifest: plugin.Manifest{ + ID: browserInboxPluginID, + Permissions: []string{"storage.namespace"}, + }, + Status: plugin.StatusLoaded, + Enabled: true, + }}, + } + app.ensureBrowserInboxSubscriptions() + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + if err := app.storage.WritePluginSetting(browserInboxPluginID, "domainBindings", map[string]interface{}{ + "example.com": "Project", + }); err != nil { + t.Errorf("WritePluginSetting(domainBindings): %v", err) + } + }() + for i := 0; i < 50; i++ { + wg.Add(1) + go func(index int) { + defer wg.Done() + if err := app.recordBrowserCapture(events.Event{ + Name: "browser.capture.page", + Timestamp: "2026-07-11T12:00:00Z", + Payload: map[string]interface{}{ + "captureId": fmt.Sprintf("capture-%02d", index), + "kind": "page", + "url": fmt.Sprintf("https://example.com/%d", index), + }, + }); err != nil { + t.Errorf("RecordBrowserCapture(%d): %v", index, err) + } + }(i) + } + wg.Wait() + + bus.Publish(events.Event{ + Name: browserInboxMutationEvent, + Payload: map[string]interface{}{ + "pluginId": browserInboxPluginID, + "action": "assign", + "captureId": "capture-00", + "workspaceRootPath": "Project", + }, + }) + bus.Publish(events.Event{ + Name: browserInboxMutationEvent, + Payload: map[string]interface{}{ + "pluginId": browserInboxPluginID, + "action": "delete", + "captureId": "capture-01", + }, + }) + + settings, err := app.storage.ReadPluginSettings(browserInboxPluginID) + if err != nil { + t.Fatalf("ReadPluginSettings: %v", err) + } + stored, ok := settings[browserInboxGlobalKey].([]interface{}) + if !ok || len(stored) != 49 { + t.Fatalf("captures:global contains %d captures, want 49", len(stored)) + } + var assigned map[string]interface{} + for _, item := range stored { + capture := item.(map[string]interface{}) + if capture["captureId"] == "capture-00" { + assigned = capture + } + if capture["captureId"] == "capture-01" { + t.Fatal("deleted capture remained in storage") + } + } + if assigned == nil || assigned["workspaceRootPath"] != "Project" { + t.Fatalf("assigned capture = %#v", assigned) + } + if settings["domainBindings"] == nil { + t.Fatal("concurrent domainBindings update was lost") + } +} + +func TestBrowserInboxMutationMigratesAndClearsLegacyCaptures(t *testing.T) { + v := vault.NewVault(nil) + if err := v.CreateVault(t.TempDir()); err != nil { + t.Fatalf("CreateVault: %v", err) + } + bus := events.NewBus() + app := &App{ + eventBus: bus, + storage: storage.New(v), + vault: v, + plugins: []plugin.Plugin{{ + Manifest: plugin.Manifest{ID: browserInboxPluginID, Permissions: []string{"storage.namespace"}}, + Status: plugin.StatusLoaded, + Enabled: true, + }}, + } + if err := app.storage.WritePluginSettings(browserInboxPluginID, map[string]interface{}{ + browserInboxLegacyKey: []interface{}{map[string]interface{}{ + "captureId": "legacy-duplicate", + "title": "Stale legacy copy", + }}, + browserInboxGlobalKey: []interface{}{map[string]interface{}{ + "captureId": "legacy-duplicate", + "title": "Canonical copy", + "workspaceRootPath": "Project", + "processed": true, + }}, + }); err != nil { + t.Fatalf("WritePluginSettings: %v", err) + } + app.ensureBrowserInboxSubscriptions() + + bus.Publish(events.Event{Name: browserInboxMutationEvent, Payload: map[string]interface{}{ + "pluginId": browserInboxPluginID, + "action": "migrate", + }}) + + settings, err := app.storage.ReadPluginSettings(browserInboxPluginID) + if err != nil { + t.Fatalf("ReadPluginSettings: %v", err) + } + legacy := settings[browserInboxLegacyKey].([]interface{}) + if len(legacy) != 0 { + t.Fatalf("legacy captures = %#v, want empty", legacy) + } + stored := settings[browserInboxGlobalKey].([]interface{}) + if len(stored) != 1 { + t.Fatalf("canonical captures = %#v, want one", stored) + } + capture := stored[0].(map[string]interface{}) + if capture["title"] != "Canonical copy" || capture["workspaceRootPath"] != "Project" || capture["processed"] != true { + t.Fatalf("canonical capture = %#v", capture) + } +} + func TestActivityFromEventRedactsBinaryPayload(t *testing.T) { activity := activityFromEvent(events.Event{ Name: "browser.capture.file", diff --git a/internal/core/browserreceiver/receiver.go b/internal/core/browserreceiver/receiver.go index 3be6a1b..68e82c8 100644 --- a/internal/core/browserreceiver/receiver.go +++ b/internal/core/browserreceiver/receiver.go @@ -57,6 +57,8 @@ type WorkspaceProvider func() string type Options struct { RequireToken bool ReceiverToken string + Available func() bool + Persist func(events.Event) error } type Server struct { @@ -128,6 +130,17 @@ func (r *Receiver) SetReceiverToken(token string) { r.options.ReceiverToken = strings.TrimSpace(token) } +// SetPersistence configures durable capture storage and its availability gate. +func (r *Receiver) SetPersistence(available func() bool, persist func(events.Event) error) { + if r == nil { + return + } + r.optionsMu.Lock() + defer r.optionsMu.Unlock() + r.options.Available = available + r.options.Persist = persist +} + func Start(addr string, receiver *Receiver) (*Server, error) { if receiver == nil { return nil, fmt.Errorf("receiver is required") @@ -181,6 +194,11 @@ func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) { writeError(w, http.StatusUnauthorized, err.Error()) return } + options := r.currentOptions() + if options.Available != nil && !options.Available() { + writeError(w, http.StatusServiceUnavailable, "browser inbox unavailable") + return + } defer req.Body.Close() decoder := json.NewDecoder(http.MaxBytesReader(w, req.Body, maxCaptureBodyBytes)) @@ -209,17 +227,27 @@ func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) { } eventName := "browser.capture." + payload.Kind - if r.bus == nil || !r.bus.HasSubscribers(eventName) { + if options.Persist == nil && (r.bus == nil || !r.bus.HasSubscribers(eventName)) { writeError(w, http.StatusServiceUnavailable, "browser inbox unavailable") return } eventPayload := payload.EventPayload() r.annotateWorkspace(eventPayload) - r.bus.Publish(events.Event{ + event := events.Event{ Name: eventName, Timestamp: time.Now().UTC().Format(time.RFC3339Nano), Payload: eventPayload, - }) + } + if options.Persist != nil { + if err := options.Persist(event); err != nil { + log.Printf("[browserreceiver] persist %s: %v", payload.CaptureID, err) + writeError(w, http.StatusServiceUnavailable, "browser inbox unavailable") + return + } + } + if r.bus != nil { + r.bus.Publish(event) + } w.WriteHeader(http.StatusAccepted) _ = json.NewEncoder(w).Encode(map[string]string{ @@ -228,6 +256,15 @@ func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) { }) } +func (r *Receiver) currentOptions() Options { + if r == nil { + return Options{} + } + r.optionsMu.RLock() + defer r.optionsMu.RUnlock() + return r.options +} + func (r *Receiver) validateReceiverToken(req *http.Request) error { if r == nil { return nil diff --git a/internal/core/browserreceiver/receiver_test.go b/internal/core/browserreceiver/receiver_test.go index 60c2342..3d10284 100644 --- a/internal/core/browserreceiver/receiver_test.go +++ b/internal/core/browserreceiver/receiver_test.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/base64" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -321,6 +322,61 @@ func TestReceiverAcceptsPairedToken(t *testing.T) { } } +func TestReceiverRejectsCaptureWhenInboxIsUnavailable(t *testing.T) { + bus := events.NewBus() + bus.Subscribe("browser.capture.page", func(event events.Event) { + t.Fatalf("unexpected event while inbox is unavailable: %#v", event) + }) + receiver := NewWithOptions(bus, Options{ + Available: func() bool { return false }, + }) + req := httptest.NewRequest(http.MethodPost, capturePath, strings.NewReader(`{ + "schemaVersion": 1, + "captureId": "capture-no-vault", + "capturedAt": "2026-07-11T12:00:00Z", + "kind": "page", + "page": {"url": "https://example.com"} + }`)) + res := httptest.NewRecorder() + + receiver.ServeHTTP(res, req) + + if res.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want %d; body=%s", res.Code, http.StatusServiceUnavailable, res.Body.String()) + } +} + +func TestReceiverDoesNotAcknowledgeFailedPersistence(t *testing.T) { + bus := events.NewBus() + published := false + bus.Subscribe("browser.capture.page", func(event events.Event) { + published = true + }) + receiver := NewWithOptions(bus, Options{ + Available: func() bool { return true }, + Persist: func(event events.Event) error { + return errors.New("disk full") + }, + }) + req := httptest.NewRequest(http.MethodPost, capturePath, strings.NewReader(`{ + "schemaVersion": 1, + "captureId": "capture-disk-full", + "capturedAt": "2026-07-11T12:00:00Z", + "kind": "page", + "page": {"url": "https://example.com"} + }`)) + res := httptest.NewRecorder() + + receiver.ServeHTTP(res, req) + + if res.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want %d; body=%s", res.Code, http.StatusServiceUnavailable, res.Body.String()) + } + if published { + t.Fatal("capture event was published after persistence failed") + } +} + func TestReceiverRotatesPairedToken(t *testing.T) { bus := events.NewBus() bus.Subscribe("browser.capture.page", func(event events.Event) {}) diff --git a/internal/core/storage/api.go b/internal/core/storage/api.go index 3cce7c7..07a6872 100644 --- a/internal/core/storage/api.go +++ b/internal/core/storage/api.go @@ -84,13 +84,22 @@ func atomicWrite(path string, data []byte) error { // ReadPluginSettings reads all settings for a plugin. // Returns empty map if settings.json does not exist. func (s *Storage) ReadPluginSettings(pluginID string) (map[string]interface{}, error) { + s.mu.RLock() + defer s.mu.RUnlock() if err := validatePluginID(pluginID); err != nil { return nil, err } + var result map[string]interface{} + err := s.withOpenVault(func(vaultPath string) error { + var err error + result, err = readPluginSettingsAt(vaultPath, pluginID) + return err + }) + return result, err +} - dir := s.vault.GetPluginSettingsPath(pluginID) - path := filepath.Join(dir, "settings.json") - +func readPluginSettingsAt(vaultPath, pluginID string) (map[string]interface{}, error) { + path := filepath.Join(vaultPath, ".verstak", "plugin-settings", pluginID, "settings.json") data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { @@ -108,13 +117,18 @@ func (s *Storage) ReadPluginSettings(pluginID string) (map[string]interface{}, e // WritePluginSettings writes all settings for a plugin atomically. func (s *Storage) WritePluginSettings(pluginID string, data map[string]interface{}) error { + s.mu.Lock() + defer s.mu.Unlock() if err := validatePluginID(pluginID); err != nil { return err } + return s.withOpenVault(func(vaultPath string) error { + return writePluginSettingsAt(vaultPath, pluginID, data) + }) +} - dir := s.vault.GetPluginSettingsPath(pluginID) - path := filepath.Join(dir, "settings.json") - +func writePluginSettingsAt(vaultPath, pluginID string, data map[string]interface{}) error { + path := filepath.Join(vaultPath, ".verstak", "plugin-settings", pluginID, "settings.json") encoded, err := json.MarshalIndent(data, "", " ") if err != nil { return fmt.Errorf("failed to marshal settings for plugin %s: %w", pluginID, err) @@ -137,12 +151,39 @@ func (s *Storage) ReadPluginSetting(pluginID, key string) (interface{}, error) { // WritePluginSetting writes a single setting key. func (s *Storage) WritePluginSetting(pluginID, key string, value interface{}) error { - settings, err := s.ReadPluginSettings(pluginID) - if err != nil { + return s.UpdatePluginSettings(pluginID, func(settings map[string]interface{}) error { + settings[key] = value + return nil + }) +} + +// UpdatePluginSettings atomically reads, mutates, and writes a plugin's settings. +func (s *Storage) UpdatePluginSettings(pluginID string, update func(map[string]interface{}) error) error { + if update == nil { + return fmt.Errorf("settings update is nil") + } + s.mu.Lock() + defer s.mu.Unlock() + if err := validatePluginID(pluginID); err != nil { return err } - settings[key] = value - return s.WritePluginSettings(pluginID, settings) + return s.withOpenVault(func(vaultPath string) error { + settings, err := readPluginSettingsAt(vaultPath, pluginID) + if err != nil { + return err + } + if err := update(settings); err != nil { + return err + } + return writePluginSettingsAt(vaultPath, pluginID, settings) + }) +} + +func (s *Storage) withOpenVault(operation func(string) error) error { + if s == nil || s.vault == nil { + return fmt.Errorf("vault is not initialized") + } + return s.vault.WithOpenPath(operation) } // ─── Data JSON API ──────────────────────────────────────── diff --git a/internal/core/vault/vault.go b/internal/core/vault/vault.go index 08f48bd..cb3fd22 100644 --- a/internal/core/vault/vault.go +++ b/internal/core/vault/vault.go @@ -70,6 +70,23 @@ func (v *Vault) GetVaultPath() string { return v.path } +// WithOpenPath runs fn while holding a read lease on the currently open vault. +// CloseVault and OpenVault wait until the operation completes. +func (v *Vault) WithOpenPath(fn func(string) error) error { + if v == nil { + return fmt.Errorf("vault is not initialized") + } + if fn == nil { + return fmt.Errorf("vault path operation is nil") + } + v.mu.RLock() + defer v.mu.RUnlock() + if v.status != StatusOpen || strings.TrimSpace(v.path) == "" { + return fmt.Errorf("vault is not open") + } + return fn(v.path) +} + // GetVaultMeta returns the current vault metadata. func (v *Vault) GetVaultMeta() *VaultMeta { v.mu.RLock()