diff --git a/frontend/src/lib/plugin-host/VerstakPluginAPI.js b/frontend/src/lib/plugin-host/VerstakPluginAPI.js index d5f56f7..d730752 100644 --- a/frontend/src/lib/plugin-host/VerstakPluginAPI.js +++ b/frontend/src/lib/plugin-host/VerstakPluginAPI.js @@ -262,6 +262,26 @@ export function createPluginAPI(pluginId) { return data || {}; }); }, + readNDJSON: function(name) { + assertActive('storage.data.readNDJSON(' + name + ')'); + if (!name) { + throw new Error('storage.data.readNDJSON requires a name'); + } + return callBackend(pluginId, 'storage.data.readNDJSON(' + name + ')', function() { + return App.ReadPluginDataNDJSON(pluginId, name); + }).then(function(records) { + return Array.isArray(records) ? records : []; + }); + }, + writeNDJSON: function(name, records) { + assertActive('storage.data.writeNDJSON(' + name + ')'); + if (!name) { + throw new Error('storage.data.writeNDJSON requires a name'); + } + return callBackendErrorString(pluginId, 'storage.data.writeNDJSON(' + name + ')', function() { + return App.WritePluginDataNDJSON(pluginId, name, Array.isArray(records) ? records : []); + }); + }, write: function(name, data) { assertActive('storage.data.write(' + name + ')'); if (!name) { diff --git a/frontend/src/lib/test/wails-mock.js b/frontend/src/lib/test/wails-mock.js index 526fbbc..df19016 100644 --- a/frontend/src/lib/test/wails-mock.js +++ b/frontend/src/lib/test/wails-mock.js @@ -3341,11 +3341,20 @@ var data = (pluginData[pluginId] && pluginData[pluginId][name]) || {}; return Promise.resolve([Object.assign({}, data), '']); }, + ReadPluginDataNDJSON: function (pluginId, name) { + var data = (pluginData[pluginId] && pluginData[pluginId][name]) || []; + return Promise.resolve([Array.isArray(data) ? data.slice() : [], '']); + }, WritePluginDataJSON: function (pluginId, name, data) { pluginData[pluginId] = pluginData[pluginId] || {}; pluginData[pluginId][name] = Object.assign({}, data || {}); return Promise.resolve(''); }, + WritePluginDataNDJSON: function (pluginId, name, records) { + pluginData[pluginId] = pluginData[pluginId] || {}; + pluginData[pluginId][name] = Array.isArray(records) ? records.slice() : []; + return Promise.resolve(''); + }, OpenWorkbenchResource: function (pluginId, request) { return openWorkbenchResource(pluginId, request || {}, ''); }, diff --git a/frontend/wailsjs/go/api/App.d.ts b/frontend/wailsjs/go/api/App.d.ts index 5e8465d..2284e77 100755 --- a/frontend/wailsjs/go/api/App.d.ts +++ b/frontend/wailsjs/go/api/App.d.ts @@ -124,6 +124,8 @@ export function PublishPluginEvent(arg1:string,arg2:string,arg3:Record>; +export function ReadPluginDataNDJSON(arg1:string,arg2:string):Promise>>; + export function ReadPluginSetting(arg1:string,arg2:string):Promise; export function ReadPluginSettings(arg1:string):Promise|string>; @@ -176,6 +178,8 @@ export function WriteFrontendLog(arg1:string,arg2:string):Promise; export function WritePluginDataJSON(arg1:string,arg2:string,arg3:Record):Promise; +export function WritePluginDataNDJSON(arg1:string,arg2:string,arg3:Array>):Promise; + export function WritePluginSetting(arg1:string,arg2:string,arg3:any):Promise; export function WritePluginSettings(arg1:string,arg2:Record):Promise; diff --git a/frontend/wailsjs/go/api/App.js b/frontend/wailsjs/go/api/App.js index 49c29e0..1976fb1 100755 --- a/frontend/wailsjs/go/api/App.js +++ b/frontend/wailsjs/go/api/App.js @@ -234,6 +234,10 @@ export function ReadPluginDataJSON(arg1, arg2) { return window['go']['api']['App']['ReadPluginDataJSON'](arg1, arg2); } +export function ReadPluginDataNDJSON(arg1, arg2) { + return window['go']['api']['App']['ReadPluginDataNDJSON'](arg1, arg2); +} + export function ReadPluginSetting(arg1, arg2) { return window['go']['api']['App']['ReadPluginSetting'](arg1, arg2); } @@ -338,6 +342,10 @@ export function WritePluginDataJSON(arg1, arg2, arg3) { return window['go']['api']['App']['WritePluginDataJSON'](arg1, arg2, arg3); } +export function WritePluginDataNDJSON(arg1, arg2, arg3) { + return window['go']['api']['App']['WritePluginDataNDJSON'](arg1, arg2, arg3); +} + export function WritePluginSetting(arg1, arg2, arg3) { return window['go']['api']['App']['WritePluginSetting'](arg1, arg2, arg3); } diff --git a/internal/api/app.go b/internal/api/app.go index 3bb1c12..328b415 100644 --- a/internal/api/app.go +++ b/internal/api/app.go @@ -40,9 +40,11 @@ var newSyncClient = syncsvc.NewClient var emitFrontendEvent = runtime.EventsEmit const pluginEventRuntimeName = "verstak:plugin-event" -const activityGlobalKey = "events:global" -const activityWorkspacePrefix = "events:workspace:" -const maxActivityEvents = 250 +const activityPluginID = "verstak.activity" +const activityRawDataName = "activity-events" +const maxActivityRawEvents = 10000 +const maxActivityRawBytes = 8 * 1024 * 1024 +const activityRetention = 60 * 24 * time.Hour const browserInboxPluginID = "verstak.browser-inbox" const browserInboxGlobalKey = "captures:global" const browserInboxLegacyKey = "captures" @@ -132,6 +134,7 @@ func NewApp( app.ensureBrowserInboxSubscriptions() if app.browserReceiver != nil { app.browserReceiver.SetPersistence(app.browserInboxAvailable, app.recordBrowserCapture) + app.browserReceiver.SetActivityPersistence(app.activityAvailable, app.recordBrowserActivityBatch) } app.startFileWatcherForOpenVault() return app @@ -205,6 +208,78 @@ func (a *App) browserInboxAvailable() bool { return true } +func (a *App) activityAvailable() bool { + if a == nil || a.storage == nil || a.vault == nil || a.vault.GetVaultStatus() != vault.StatusOpen { + return false + } + _, err := a.requirePluginAccess(activityPluginID, "storage.namespace") + return err == nil +} + +func (a *App) recordBrowserActivityBatch(event events.Event) error { + if !a.activityAvailable() { + return fmt.Errorf("activity storage unavailable") + } + payload := eventPayloadMap(event.Payload) + batchID := firstPayloadText(payload, "batchId") + if batchID == "" { + return fmt.Errorf("batchId is empty") + } + entries, ok := payload["entries"].([]map[string]interface{}) + if !ok || len(entries) == 0 { + return fmt.Errorf("activity batch entries are empty") + } + now := time.Now().UTC().Format(time.RFC3339Nano) + receivedAt := event.Timestamp + if receivedAt == "" { + receivedAt = now + } + records := make([]map[string]interface{}, 0, len(entries)) + for index, entry := range entries { + hostname := firstPayloadText(entry, "hostname") + endedAt := firstPayloadText(entry, "endedAt") + if hostname == "" || endedAt == "" { + return fmt.Errorf("activity batch entry %d is invalid", index) + } + durationSeconds, _ := entry["durationSeconds"].(int64) + if durationSeconds == 0 { + if number, ok := entry["durationSeconds"].(float64); ok { + durationSeconds = int64(number) + } + } + records = append(records, map[string]interface{}{ + "activityId": fmt.Sprintf("browser-domain:%s:%d", batchID, index), + "type": "browser.activity.domain", + "title": hostname, + "summary": fmt.Sprintf("%d min browser activity", durationSeconds/60), + "occurredAt": endedAt, + "receivedAt": receivedAt, + "sourcePluginId": "verstak-browser-extension", + "sourceBatchId": batchID, + "hostname": hostname, + "startedAt": firstPayloadText(entry, "startedAt"), + "endedAt": endedAt, + "durationSeconds": durationSeconds, + "workspaceRootPath": "", + "payload": map[string]interface{}{ + "hostname": hostname, + "startedAt": firstPayloadText(entry, "startedAt"), + "endedAt": endedAt, + "durationSeconds": durationSeconds, + }, + }) + } + _, err := a.storage.AppendPluginDataNDJSON(activityPluginID, activityRawDataName, records, storage.NDJSONRetention{ + TimestampField: "occurredAt", + MaxAge: activityRetention, + MaxEntries: maxActivityRawEvents, + MaxBytes: maxActivityRawBytes, + DeduplicateField: "sourceBatchId", + DeduplicateValue: batchID, + }) + return err +} + func (a *App) recordBrowserCapture(event events.Event) error { if !a.browserInboxAvailable() { return fmt.Errorf("browser inbox unavailable") @@ -428,28 +503,13 @@ func (a *App) recordActivityProviderEvent(event events.Event) { } func (a *App) appendActivityEvent(pluginID string, activity map[string]interface{}) error { - settings, err := a.storage.ReadPluginSettings(pluginID) - if err != nil { - return err - } - key := activityGlobalKey - if workspace, _ := activity["workspaceRootPath"].(string); strings.TrimSpace(workspace) != "" { - key = activityWorkspacePrefix + url.QueryEscape(strings.TrimSpace(workspace)) - key = strings.ReplaceAll(key, "+", "%20") - } - eventsList := []interface{}{activity} - if existing, ok := settings[key].([]interface{}); ok { - eventsList = append(eventsList, existing...) - } else if existingMaps, ok := settings[key].([]map[string]interface{}); ok { - for _, item := range existingMaps { - eventsList = append(eventsList, item) - } - } - if len(eventsList) > maxActivityEvents { - eventsList = eventsList[:maxActivityEvents] - } - settings[key] = eventsList - return a.storage.WritePluginSettings(pluginID, settings) + _, err := a.storage.AppendPluginDataNDJSON(pluginID, activityRawDataName, []map[string]interface{}{activity}, storage.NDJSONRetention{ + TimestampField: "occurredAt", + MaxAge: activityRetention, + MaxEntries: maxActivityRawEvents, + MaxBytes: maxActivityRawBytes, + }) + return err } func activityFromEvent(event events.Event) map[string]interface{} { @@ -1016,6 +1076,40 @@ func (a *App) ReadPluginDataJSON(pluginID, name string) map[string]interface{} { return data } +// ReadPluginDataNDJSON reads append-only plugin data without exposing the +// underlying vault path to plugin frontends. +func (a *App) ReadPluginDataNDJSON(pluginID, name string) []map[string]interface{} { + if _, err := a.requirePluginAccess(pluginID, "storage.namespace"); err != nil { + log.Printf("[api] ReadPluginDataNDJSON(%s, %s): %v", pluginID, name, err) + return []map[string]interface{}{} + } + if a.storage == nil { + return []map[string]interface{}{} + } + data, err := a.storage.ReadPluginDataNDJSON(pluginID, name) + if err != nil { + log.Printf("[api] ReadPluginDataNDJSON(%s, %s): %v", pluginID, name, err) + return []map[string]interface{}{} + } + return data +} + +// WritePluginDataNDJSON replaces append-only data after an explicit user +// action, such as clearing activity history. +func (a *App) WritePluginDataNDJSON(pluginID, name string, data []map[string]interface{}) string { + if _, err := a.requirePluginAccess(pluginID, "storage.namespace"); err != nil { + return err.Error() + } + if a.storage == nil { + return "storage not initialized" + } + if err := a.storage.WritePluginDataNDJSON(pluginID, name, data); err != nil { + log.Printf("[api] WritePluginDataNDJSON(%s, %s): %v", pluginID, name, err) + return err.Error() + } + return "" +} + // WritePluginDataJSON writes a named JSON data file for a plugin. func (a *App) WritePluginDataJSON(pluginID, name string, data map[string]interface{}) string { if _, err := a.requirePluginAccess(pluginID, "storage.namespace"); err != nil { diff --git a/internal/api/app_test.go b/internal/api/app_test.go index 21e23a2..80fe32b 100644 --- a/internal/api/app_test.go +++ b/internal/api/app_test.go @@ -593,21 +593,14 @@ func TestActivityProviderRecordsFileChangedWithoutMountedView(t *testing.T) { t.Fatalf("WriteVaultTextFile: %s", errStr) } - settings, err := app.storage.ReadPluginSettings("verstak.activity") + stored, err := app.storage.ReadPluginDataNDJSON("verstak.activity", activityRawDataName) if err != nil { - t.Fatalf("ReadPluginSettings: %v", err) - } - stored, ok := settings["events:workspace:Project"].([]interface{}) - if !ok { - t.Fatalf("events:workspace:Project = %#v, want []interface{}", settings["events:workspace:Project"]) + t.Fatalf("ReadPluginDataNDJSON: %v", err) } if len(stored) != 1 { t.Fatalf("stored %d activity events, want 1", len(stored)) } - activity, ok := stored[0].(map[string]interface{}) - if !ok { - t.Fatalf("activity = %#v, want map[string]interface{}", stored[0]) - } + activity := stored[0] if activity["type"] != "file.changed" { t.Fatalf("activity type = %#v, want file.changed", activity["type"]) } @@ -619,6 +612,51 @@ func TestActivityProviderRecordsFileChangedWithoutMountedView(t *testing.T) { } } +func TestBrowserActivityBatchPersistsWithoutMountedViewAndDeduplicates(t *testing.T) { + v := vault.NewVault(nil) + if err := v.CreateVault(t.TempDir()); err != nil { + t.Fatalf("CreateVault: %v", err) + } + app := &App{ + storage: storage.New(v), + vault: v, + plugins: []plugin.Plugin{{ + Manifest: plugin.Manifest{ID: activityPluginID, Permissions: []string{"storage.namespace"}}, + Status: plugin.StatusLoaded, + Enabled: true, + }}, + } + event := events.Event{ + Name: "browser.activity.batch", + Timestamp: "2026-07-12T10:05:01Z", + Payload: map[string]interface{}{ + "batchId": "batch-activity-1", + "entries": []map[string]interface{}{{ + "hostname": "example.com", "startedAt": "2026-07-12T10:00:00Z", "endedAt": "2026-07-12T10:05:00Z", "durationSeconds": int64(300), + }}, + }, + } + if err := app.recordBrowserActivityBatch(event); err != nil { + t.Fatalf("recordBrowserActivityBatch: %v", err) + } + if err := app.recordBrowserActivityBatch(event); err != nil { + t.Fatalf("recordBrowserActivityBatch retry: %v", err) + } + records, err := app.storage.ReadPluginDataNDJSON(activityPluginID, activityRawDataName) + if err != nil { + t.Fatalf("ReadPluginDataNDJSON: %v", err) + } + if len(records) != 1 { + t.Fatalf("records = %+v, want one idempotent domain activity", records) + } + if records[0]["type"] != "browser.activity.domain" || records[0]["hostname"] != "example.com" { + t.Fatalf("record = %+v, want domain-only activity", records[0]) + } + if _, ok := records[0]["url"]; ok { + t.Fatalf("record must not contain URL: %+v", records[0]) + } +} + func TestBrowserInboxRecordsCaptureWithoutMountedView(t *testing.T) { v := vault.NewVault(nil) if err := v.CreateVault(t.TempDir()); err != nil { diff --git a/internal/core/browserreceiver/receiver.go b/internal/core/browserreceiver/receiver.go index 68e82c8..f3b12e0 100644 --- a/internal/core/browserreceiver/receiver.go +++ b/internal/core/browserreceiver/receiver.go @@ -12,18 +12,20 @@ import ( "log" "net" "net/http" - "net/url" "strings" "sync" "time" "github.com/verstak/verstak-desktop/internal/core/events" + "github.com/verstak/verstak-desktop/internal/core/hostname" ) const ( capturePath = "/api/browser-inbox/v1/captures" + activityBatchPath = "/api/browser-activity/v1/batches" DefaultAddr = "127.0.0.1:47731" DefaultCaptureURL = "http://" + DefaultAddr + capturePath + DefaultActivityURL = "http://" + DefaultAddr + activityBatchPath receiverTokenHeader = "X-Verstak-Receiver-Token" ) @@ -43,6 +45,9 @@ const ( maxFileTextBytes = 2 * 1024 * 1024 maxFileBytes = 8 * 1024 * 1024 maxFileDataBase64Bytes = 4 * ((maxFileBytes + 2) / 3) + maxActivityBodyBytes = 256 * 1024 + maxActivityEntries = 100 + maxActivityDuration = 10 * time.Minute ) type Receiver struct { @@ -55,10 +60,12 @@ type Receiver struct { type WorkspaceProvider func() string type Options struct { - RequireToken bool - ReceiverToken string - Available func() bool - Persist func(events.Event) error + RequireToken bool + ReceiverToken string + Available func() bool + Persist func(events.Event) error + ActivityAvailable func() bool + PersistActivity func(events.Event) error } type Server struct { @@ -107,6 +114,23 @@ type CaptureBrowser struct { Name string `json:"name"` } +// ActivityBatchPayload contains only domain-level time accounting. It never +// accepts or emits page URLs, titles, content, or navigation history. +type ActivityBatchPayload struct { + SchemaVersion int `json:"schemaVersion"` + BatchID string `json:"batchId"` + CreatedAt string `json:"createdAt"` + Source string `json:"source"` + Entries []ActivityEntry `json:"entries"` +} + +type ActivityEntry struct { + Hostname string `json:"hostname"` + StartedAt string `json:"startedAt"` + EndedAt string `json:"endedAt"` + DurationSeconds int64 `json:"durationSeconds"` +} + func New(bus *events.Bus, providers ...WorkspaceProvider) *Receiver { return NewWithOptions(bus, Options{}, providers...) } @@ -141,6 +165,18 @@ func (r *Receiver) SetPersistence(available func() bool, persist func(events.Eve r.options.Persist = persist } +// SetActivityPersistence configures durable passive browser activity storage. +// An acknowledgement is sent only after persist returns successfully. +func (r *Receiver) SetActivityPersistence(available func() bool, persist func(events.Event) error) { + if r == nil { + return + } + r.optionsMu.Lock() + defer r.optionsMu.Unlock() + r.options.ActivityAvailable = available + r.options.PersistActivity = persist +} + func Start(addr string, receiver *Receiver) (*Server, error) { if receiver == nil { return nil, fmt.Errorf("receiver is required") @@ -181,6 +217,10 @@ func (s *Server) Close() error { func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) { w.Header().Set("Content-Type", "application/json") + if req.URL.Path == activityBatchPath { + r.serveActivityBatch(w, req) + return + } if req.URL.Path != capturePath { http.NotFound(w, req) return @@ -256,6 +296,59 @@ func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) { }) } +func (r *Receiver) serveActivityBatch(w http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodPost { + w.WriteHeader(http.StatusMethodNotAllowed) + _ = json.NewEncoder(w).Encode(map[string]string{"error": "method not allowed"}) + return + } + if err := r.validateReceiverToken(req); err != nil { + writeError(w, http.StatusUnauthorized, err.Error()) + return + } + options := r.currentOptions() + if options.ActivityAvailable == nil || !options.ActivityAvailable() || options.PersistActivity == nil { + writeError(w, http.StatusServiceUnavailable, "activity storage unavailable") + return + } + + defer req.Body.Close() + decoder := json.NewDecoder(http.MaxBytesReader(w, req.Body, maxActivityBodyBytes)) + var batch ActivityBatchPayload + if err := decoder.Decode(&batch); err != nil { + writeActivityDecodeError(w, err) + return + } + if err := decoder.Decode(&struct{}{}); err != io.EOF { + writeActivityDecodeError(w, err) + return + } + if err := batch.NormalizeAndValidate(); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + event := events.Event{ + Name: "browser.activity.batch", + Timestamp: time.Now().UTC().Format(time.RFC3339Nano), + Payload: batch.EventPayload(), + } + if err := options.PersistActivity(event); err != nil { + log.Printf("[browserreceiver] persist activity %s: %v", batch.BatchID, err) + writeError(w, http.StatusServiceUnavailable, "activity storage unavailable") + return + } + if r.bus != nil { + r.bus.Publish(event) + } + + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(map[string]string{ + "status": "accepted", + "batchId": batch.BatchID, + }) +} + func (r *Receiver) currentOptions() Options { if r == nil { return Options{} @@ -370,6 +463,80 @@ func (p CapturePayload) Validate() error { return nil } +// NormalizeAndValidate validates a batch before it is made durable and replaces +// each submitted hostname with the shared canonical A-label form. +func (p *ActivityBatchPayload) NormalizeAndValidate() error { + if p == nil || p.SchemaVersion != 1 { + return fmt.Errorf("unsupported schemaVersion") + } + if strings.TrimSpace(p.BatchID) == "" { + return fmt.Errorf("batchId is required") + } + if err := validateCaptureText(p.BatchID, "batchId", maxCaptureIDBytes); err != nil { + return err + } + if strings.TrimSpace(p.CreatedAt) == "" { + return fmt.Errorf("createdAt is required") + } + if _, err := time.Parse(time.RFC3339, p.CreatedAt); err != nil { + return fmt.Errorf("createdAt is invalid") + } + if err := validateCaptureText(p.CreatedAt, "createdAt", maxCapturedAtBytes); err != nil { + return err + } + if strings.TrimSpace(p.Source) == "" { + return fmt.Errorf("source is required") + } + if err := validateCaptureText(p.Source, "source", maxCaptureSourceBytes); err != nil { + return err + } + if len(p.Entries) == 0 || len(p.Entries) > maxActivityEntries { + return fmt.Errorf("entries must contain between 1 and %d items", maxActivityEntries) + } + for index := range p.Entries { + entry := &p.Entries[index] + canonical := hostname.NormalizeHostnameV1(entry.Hostname) + if canonical == "" { + return fmt.Errorf("entries[%d].hostname is invalid", index) + } + startedAt, err := time.Parse(time.RFC3339, entry.StartedAt) + if err != nil { + return fmt.Errorf("entries[%d].startedAt is invalid", index) + } + endedAt, err := time.Parse(time.RFC3339, entry.EndedAt) + if err != nil { + return fmt.Errorf("entries[%d].endedAt is invalid", index) + } + interval := endedAt.Sub(startedAt) + if interval <= 0 || interval > maxActivityDuration { + return fmt.Errorf("entries[%d] interval must be between 1 second and %s", index, maxActivityDuration) + } + if entry.DurationSeconds <= 0 || entry.DurationSeconds > int64(maxActivityDuration/time.Second) || time.Duration(entry.DurationSeconds)*time.Second > interval { + return fmt.Errorf("entries[%d].durationSeconds is invalid", index) + } + entry.Hostname = canonical + } + return nil +} + +func (p ActivityBatchPayload) EventPayload() map[string]interface{} { + entries := make([]map[string]interface{}, 0, len(p.Entries)) + for _, entry := range p.Entries { + entries = append(entries, map[string]interface{}{ + "hostname": entry.Hostname, + "startedAt": entry.StartedAt, + "endedAt": entry.EndedAt, + "durationSeconds": entry.DurationSeconds, + }) + } + return map[string]interface{}{ + "batchId": strings.TrimSpace(p.BatchID), + "createdAt": strings.TrimSpace(p.CreatedAt), + "source": strings.TrimSpace(p.Source), + "entries": entries, + } +} + func (p CapturePayload) validateFile() error { if p.File == nil || strings.TrimSpace(p.File.Name) == "" { return fmt.Errorf("file.name is required") @@ -450,10 +617,19 @@ func (p CapturePayload) EventPayload() map[string]interface{} { } func captureDomain(rawURL, fallback string) string { - if u, err := url.Parse(strings.TrimSpace(rawURL)); err == nil && u.Hostname() != "" { - return u.Hostname() + if normalized := hostname.NormalizeURLHostnameV1(rawURL); normalized != "" { + return normalized } - return strings.TrimSpace(fallback) + return hostname.NormalizeHostnameV1(fallback) +} + +func writeActivityDecodeError(w http.ResponseWriter, err error) { + var maxBytesErr *http.MaxBytesError + if errors.As(err, &maxBytesErr) { + writeError(w, http.StatusRequestEntityTooLarge, "activity payload exceeds limit") + return + } + writeError(w, http.StatusBadRequest, "invalid JSON") } func writeError(w http.ResponseWriter, status int, message string) { diff --git a/internal/core/browserreceiver/receiver_test.go b/internal/core/browserreceiver/receiver_test.go index 3d10284..c99153e 100644 --- a/internal/core/browserreceiver/receiver_test.go +++ b/internal/core/browserreceiver/receiver_test.go @@ -238,6 +238,101 @@ func TestReceiverLeavesCaptureUnassignedWithoutCurrentWorkspace(t *testing.T) { } } +func TestReceiverAcceptsDomainActivityBatchAfterDurablePersistence(t *testing.T) { + bus := events.NewBus() + received := make(chan events.Event, 1) + bus.Subscribe("browser.activity.batch", func(event events.Event) { + received <- event + }) + persisted := 0 + receiver := NewWithOptions(bus, Options{ + ActivityAvailable: func() bool { return true }, + PersistActivity: func(event events.Event) error { + persisted++ + return nil + }, + }) + body := `{ + "schemaVersion": 1, + "batchId": "batch-123", + "createdAt": "2026-07-12T10:05:00.000Z", + "source": "verstak-browser-extension", + "entries": [{ + "hostname": "пример.рф", + "startedAt": "2026-07-12T10:00:00.000Z", + "endedAt": "2026-07-12T10:05:00.000Z", + "durationSeconds": 300 + }] + }` + req := httptest.NewRequest(http.MethodPost, "/api/browser-activity/v1/batches", bytes.NewBufferString(body)) + rec := httptest.NewRecorder() + + receiver.ServeHTTP(rec, req) + + if rec.Code != http.StatusAccepted { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusAccepted, rec.Body.String()) + } + if persisted != 1 { + t.Fatalf("persisted = %d, want 1 before acknowledgement", persisted) + } + var response map[string]string + if err := json.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatalf("response json: %v", err) + } + if response["batchId"] != "batch-123" || response["status"] != "accepted" { + t.Fatalf("response = %+v, want accepted batch-123", response) + } + event := <-received + payload, ok := event.Payload.(map[string]interface{}) + if !ok { + t.Fatalf("event payload type = %T, want map[string]interface{}", event.Payload) + } + if payload["batchId"] != "batch-123" { + t.Fatalf("event payload = %+v, want batch id", payload) + } + entries, ok := payload["entries"].([]map[string]interface{}) + if !ok || len(entries) != 1 || entries[0]["hostname"] != "xn--e1afmkfd.xn--p1ai" { + t.Fatalf("event entries = %+v, want one canonical hostname", payload["entries"]) + } + if _, ok := payload["url"]; ok { + t.Fatalf("activity payload must not contain URL: %+v", payload) + } +} + +func TestReceiverRejectsDomainActivityWithoutConsumerOrValidInterval(t *testing.T) { + for _, tc := range []struct { + name string + receiver *Receiver + body string + wantCode int + }{ + { + name: "consumer unavailable", + receiver: New(events.NewBus()), + body: `{"schemaVersion":1,"batchId":"batch-a","createdAt":"2026-07-12T10:05:00Z","source":"verstak-browser-extension","entries":[{"hostname":"example.com","startedAt":"2026-07-12T10:00:00Z","endedAt":"2026-07-12T10:05:00Z","durationSeconds":300}]}`, + wantCode: http.StatusServiceUnavailable, + }, + { + name: "duration too long", + receiver: NewWithOptions(events.NewBus(), Options{ + ActivityAvailable: func() bool { return true }, + PersistActivity: func(events.Event) error { return nil }, + }), + body: `{"schemaVersion":1,"batchId":"batch-b","createdAt":"2026-07-12T10:05:00Z","source":"verstak-browser-extension","entries":[{"hostname":"example.com","startedAt":"2026-07-12T10:00:00Z","endedAt":"2026-07-12T10:20:01Z","durationSeconds":1201}]}`, + wantCode: http.StatusBadRequest, + }, + } { + t.Run(tc.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/api/browser-activity/v1/batches", strings.NewReader(tc.body)) + rec := httptest.NewRecorder() + tc.receiver.ServeHTTP(rec, req) + if rec.Code != tc.wantCode { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, tc.wantCode, rec.Body.String()) + } + }) + } +} + func TestReceiverRequiresTokenWhenPaired(t *testing.T) { bus := events.NewBus() received := make(chan events.Event, 1) diff --git a/internal/core/storage/api.go b/internal/core/storage/api.go index 07a6872..8550131 100644 --- a/internal/core/storage/api.go +++ b/internal/core/storage/api.go @@ -20,6 +20,17 @@ type Storage struct { vault *vault.Vault } +// NDJSONRetention bounds append-only plugin data. Records are compacted after +// a successful append so settings.json never becomes an event log. +type NDJSONRetention struct { + TimestampField string + MaxAge time.Duration + MaxEntries int + MaxBytes int64 + DeduplicateField string + DeduplicateValue string +} + // New creates a new Storage instance backed by the given vault. func New(v *vault.Vault) *Storage { return &Storage{vault: v} @@ -234,6 +245,204 @@ func (s *Storage) WritePluginDataJSON(pluginID, name string, data map[string]int return atomicWrite(path, encoded) } +// ReadPluginDataNDJSON reads an append-only named data file. A missing file is +// represented by an empty slice. +func (s *Storage) ReadPluginDataNDJSON(pluginID, name string) ([]map[string]interface{}, error) { + if err := validatePluginID(pluginID); err != nil { + return nil, err + } + if err := validateStorageName("data", name); err != nil { + return nil, err + } + s.mu.RLock() + defer s.mu.RUnlock() + var records []map[string]interface{} + err := s.withOpenVault(func(vaultPath string) error { + var err error + records, err = readPluginDataNDJSONAt(vaultPath, pluginID, name) + return err + }) + return records, err +} + +// AppendPluginDataNDJSON appends records durably and then applies bounded +// retention. It returns false without writing when the supplied idempotency +// value is already present in the retained log. +func (s *Storage) AppendPluginDataNDJSON(pluginID, name string, records []map[string]interface{}, retention NDJSONRetention) (bool, error) { + if err := validatePluginID(pluginID); err != nil { + return false, err + } + if err := validateStorageName("data", name); err != nil { + return false, err + } + if len(records) == 0 { + return false, fmt.Errorf("NDJSON records are empty") + } + if retention.MaxEntries < 0 || retention.MaxBytes < 0 || retention.MaxAge < 0 { + return false, fmt.Errorf("NDJSON retention values must not be negative") + } + s.mu.Lock() + defer s.mu.Unlock() + stored := false + err := s.withOpenVault(func(vaultPath string) error { + existing, err := readPluginDataNDJSONAt(vaultPath, pluginID, name) + if err != nil { + return err + } + if retention.DeduplicateField != "" && retention.DeduplicateValue != "" { + for _, record := range existing { + if fmt.Sprint(record[retention.DeduplicateField]) == retention.DeduplicateValue { + return nil + } + } + } + path := pluginDataNDJSONPath(vaultPath, pluginID, name) + if err := appendNDJSON(path, records); err != nil { + return err + } + stored = true + compacted := compactNDJSONRecords(append(existing, records...), retention, time.Now().UTC()) + if !sameNDJSONRecords(append(existing, records...), compacted) { + return writeNDJSON(path, compacted) + } + return nil + }) + return stored, err +} + +// WritePluginDataNDJSON replaces a named append-only data file. It is reserved +// for explicit user actions such as clearing Activity, never normal event +// ingestion. +func (s *Storage) WritePluginDataNDJSON(pluginID, name string, records []map[string]interface{}) error { + if err := validatePluginID(pluginID); err != nil { + return err + } + if err := validateStorageName("data", name); err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + return s.withOpenVault(func(vaultPath string) error { + return writeNDJSON(pluginDataNDJSONPath(vaultPath, pluginID, name), records) + }) +} + +func pluginDataNDJSONPath(vaultPath, pluginID, name string) string { + return filepath.Join(vaultPath, ".verstak", "plugin-data", pluginID, name+".ndjson") +} + +func readPluginDataNDJSONAt(vaultPath, pluginID, name string) ([]map[string]interface{}, error) { + data, err := os.ReadFile(pluginDataNDJSONPath(vaultPath, pluginID, name)) + if err != nil { + if os.IsNotExist(err) { + return []map[string]interface{}{}, nil + } + return nil, fmt.Errorf("failed to read NDJSON data %s for plugin %s: %w", name, pluginID, err) + } + if len(data) == 0 { + return []map[string]interface{}{}, nil + } + lines := strings.Split(strings.TrimSuffix(string(data), "\n"), "\n") + records := make([]map[string]interface{}, 0, len(lines)) + for index, line := range lines { + if strings.TrimSpace(line) == "" { + continue + } + var record map[string]interface{} + if err := json.Unmarshal([]byte(line), &record); err != nil { + return nil, fmt.Errorf("corrupt NDJSON data %s line %d for plugin %s: %w", name, index+1, pluginID, err) + } + records = append(records, record) + } + return records, nil +} + +func appendNDJSON(path string, records []map[string]interface{}) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return fmt.Errorf("failed to create NDJSON data directory: %w", err) + } + file, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return fmt.Errorf("failed to open NDJSON data file: %w", err) + } + defer file.Close() + for _, record := range records { + encoded, err := json.Marshal(record) + if err != nil { + return fmt.Errorf("failed to marshal NDJSON record: %w", err) + } + if _, err := file.Write(append(encoded, '\n')); err != nil { + return fmt.Errorf("failed to append NDJSON record: %w", err) + } + } + if err := file.Sync(); err != nil { + return fmt.Errorf("failed to sync NDJSON data: %w", err) + } + return nil +} + +func writeNDJSON(path string, records []map[string]interface{}) error { + data := make([]byte, 0) + for _, record := range records { + encoded, err := json.Marshal(record) + if err != nil { + return fmt.Errorf("failed to marshal NDJSON record: %w", err) + } + data = append(data, encoded...) + data = append(data, '\n') + } + return atomicWrite(path, data) +} + +func compactNDJSONRecords(records []map[string]interface{}, retention NDJSONRetention, now time.Time) []map[string]interface{} { + kept := make([]map[string]interface{}, 0, len(records)) + cutoff := now.Add(-retention.MaxAge) + for _, record := range records { + if retention.MaxAge > 0 && retention.TimestampField != "" { + if raw, ok := record[retention.TimestampField].(string); ok { + if timestamp, err := time.Parse(time.RFC3339, raw); err == nil && timestamp.Before(cutoff) { + continue + } + } + } + kept = append(kept, record) + } + if retention.MaxEntries > 0 && len(kept) > retention.MaxEntries { + kept = kept[len(kept)-retention.MaxEntries:] + } + if retention.MaxBytes > 0 { + for len(kept) > 0 && ndjsonSize(kept) > retention.MaxBytes { + kept = kept[1:] + } + } + return kept +} + +func ndjsonSize(records []map[string]interface{}) int64 { + var total int64 + for _, record := range records { + encoded, err := json.Marshal(record) + if err == nil { + total += int64(len(encoded) + 1) + } + } + return total +} + +func sameNDJSONRecords(left, right []map[string]interface{}) bool { + if len(left) != len(right) { + return false + } + for index := range left { + leftJSON, leftErr := json.Marshal(left[index]) + rightJSON, rightErr := json.Marshal(right[index]) + if leftErr != nil || rightErr != nil || string(leftJSON) != string(rightJSON) { + return false + } + } + return true +} + // ─── Cache JSON API ─────────────────────────────────────── // ReadPluginCacheJSON reads a named JSON cache file for a plugin. diff --git a/internal/core/storage/ndjson_test.go b/internal/core/storage/ndjson_test.go new file mode 100644 index 0000000..e50c229 --- /dev/null +++ b/internal/core/storage/ndjson_test.go @@ -0,0 +1,91 @@ +package storage + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestAppendPluginDataNDJSONCompactsAndDeduplicates(t *testing.T) { + s, vaultDir := newTestStorage(t) + retention := NDJSONRetention{ + TimestampField: "occurredAt", + MaxAge: 60 * 24 * time.Hour, + MaxEntries: 2, + MaxBytes: 16 * 1024, + DeduplicateField: "sourceBatchId", + DeduplicateValue: "batch-1", + } + old := time.Now().UTC().Add(-61 * 24 * time.Hour).Format(time.RFC3339) + now := time.Now().UTC().Format(time.RFC3339) + + stored, err := s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{ + {"activityId": "old", "sourceBatchId": "old-batch", "occurredAt": old}, + }, retention) + if err != nil || !stored { + t.Fatalf("old append = (%v, %v), want (true, nil)", stored, err) + } + + stored, err = s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{ + {"activityId": "current-1", "sourceBatchId": "batch-1", "occurredAt": now}, + }, retention) + if err != nil || !stored { + t.Fatalf("first current append = (%v, %v), want (true, nil)", stored, err) + } + + stored, err = s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{ + {"activityId": "duplicate", "sourceBatchId": "batch-1", "occurredAt": now}, + }, retention) + if err != nil || stored { + t.Fatalf("duplicate append = (%v, %v), want (false, nil)", stored, err) + } + + for _, item := range []struct { + id string + batch string + }{ + {id: "current-2", batch: "batch-3"}, + {id: "current-3", batch: "batch-4"}, + } { + retention.DeduplicateValue = item.batch + if stored, err := s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{ + {"activityId": item.id, "sourceBatchId": item.batch, "occurredAt": now}, + }, retention); err != nil || !stored { + t.Fatalf("append %s = (%v, %v), want (true, nil)", item.id, stored, err) + } + } + + records, err := s.ReadPluginDataNDJSON("verstak.activity", "activity-events") + if err != nil { + t.Fatalf("ReadPluginDataNDJSON: %v", err) + } + if len(records) != 2 || records[0]["activityId"] != "current-2" || records[1]["activityId"] != "current-3" { + t.Fatalf("records = %+v, want current-2 and current-3", records) + } + if _, err := os.Stat(filepath.Join(vaultDir, "VerstakVault", ".verstak", "plugin-data", "verstak.activity", "activity-events.ndjson")); err != nil { + t.Fatalf("activity data file missing: %v", err) + } +} + +func TestWritePluginDataNDJSONReplacesRecordsForExplicitUserClear(t *testing.T) { + s, _ := newTestStorage(t) + if _, err := s.AppendPluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{ + {"activityId": "one", "occurredAt": time.Now().UTC().Format(time.RFC3339)}, + {"activityId": "two", "occurredAt": time.Now().UTC().Format(time.RFC3339)}, + }, NDJSONRetention{}); err != nil { + t.Fatalf("AppendPluginDataNDJSON: %v", err) + } + if err := s.WritePluginDataNDJSON("verstak.activity", "activity-events", []map[string]interface{}{ + {"activityId": "two", "occurredAt": time.Now().UTC().Format(time.RFC3339)}, + }); err != nil { + t.Fatalf("WritePluginDataNDJSON: %v", err) + } + records, err := s.ReadPluginDataNDJSON("verstak.activity", "activity-events") + if err != nil { + t.Fatalf("ReadPluginDataNDJSON: %v", err) + } + if len(records) != 1 || records[0]["activityId"] != "two" { + t.Fatalf("records = %+v, want only two", records) + } +}