feat: persist handled activity session watermarks
This commit is contained in:
parent
6b556c59d9
commit
5a249dcf24
|
|
@ -42,6 +42,8 @@ var emitFrontendEvent = runtime.EventsEmit
|
|||
const pluginEventRuntimeName = "verstak:plugin-event"
|
||||
const activityPluginID = "verstak.activity"
|
||||
const activityRawDataName = "activity-events"
|
||||
const activitySessionHandlingKey = "activity-session-handling-v2"
|
||||
const activitySessionHandledEvent = "activity.session.handled"
|
||||
const maxActivityRawEvents = 10000
|
||||
const maxActivityRawBytes = 8 * 1024 * 1024
|
||||
const activityRetention = 60 * 24 * time.Hour
|
||||
|
|
@ -564,6 +566,46 @@ func (a *App) ensureActivityProviderSubscriptions() {
|
|||
})
|
||||
}
|
||||
}
|
||||
if !a.activityEvents[activitySessionHandledEvent] {
|
||||
a.activityEvents[activitySessionHandledEvent] = true
|
||||
a.eventBus.Subscribe(activitySessionHandledEvent, func(event events.Event) {
|
||||
if err := a.recordActivitySessionHandled(event); err != nil {
|
||||
log.Printf("[api] activity session handling update failed: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (a *App) recordActivitySessionHandled(event events.Event) error {
|
||||
if !a.activityAvailable() {
|
||||
return fmt.Errorf("activity storage unavailable")
|
||||
}
|
||||
payload := eventPayloadMap(event.Payload)
|
||||
if firstPayloadText(payload, "pluginId") != "verstak.journal" {
|
||||
return fmt.Errorf("activity session handling source is not authorized")
|
||||
}
|
||||
sessionID := firstPayloadText(payload, "sessionId")
|
||||
handledThrough := firstPayloadText(payload, "handledThrough")
|
||||
status := firstPayloadText(payload, "status")
|
||||
if sessionID == "" || handledThrough == "" || (status != "accepted" && status != "dismissed") {
|
||||
return fmt.Errorf("activity session handling payload is invalid")
|
||||
}
|
||||
if _, err := time.Parse(time.RFC3339, handledThrough); err != nil {
|
||||
return fmt.Errorf("activity session handledThrough is invalid")
|
||||
}
|
||||
return a.storage.UpdatePluginSettings(activityPluginID, func(settings map[string]interface{}) error {
|
||||
handled, _ := settings[activitySessionHandlingKey].(map[string]interface{})
|
||||
if handled == nil {
|
||||
handled = make(map[string]interface{})
|
||||
}
|
||||
handled[sessionID] = map[string]interface{}{
|
||||
"status": status,
|
||||
"handledThrough": handledThrough,
|
||||
"handledAt": time.Now().UTC().Format(time.RFC3339Nano),
|
||||
}
|
||||
settings[activitySessionHandlingKey] = handled
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func (a *App) recordActivityProviderEvent(event events.Event) {
|
||||
|
|
|
|||
|
|
@ -841,6 +841,37 @@ func TestOpenExternalURLUsesBrowserOpenService(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestJournalHandledSessionPersistsActivityWatermark(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,
|
||||
contribRegistry: contribution.NewRegistry(),
|
||||
plugins: []plugin.Plugin{
|
||||
{Manifest: plugin.Manifest{ID: activityPluginID, Permissions: []string{"storage.namespace"}}, Status: plugin.StatusLoaded, Enabled: true},
|
||||
{Manifest: plugin.Manifest{ID: "verstak.journal", Permissions: []string{"events.publish"}}, Status: plugin.StatusLoaded, Enabled: true},
|
||||
},
|
||||
}
|
||||
app.ensureActivityProviderSubscriptions()
|
||||
bus.Publish(events.Event{Name: activitySessionHandledEvent, Payload: map[string]interface{}{
|
||||
"pluginId": "verstak.journal", "sessionId": "session-1", "handledThrough": "2026-07-12T10:30:00Z", "status": "accepted",
|
||||
}})
|
||||
settings, err := app.storage.ReadPluginSettings(activityPluginID)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadPluginSettings: %v", err)
|
||||
}
|
||||
handled := settings[activitySessionHandlingKey].(map[string]interface{})
|
||||
record := handled["session-1"].(map[string]interface{})
|
||||
if record["status"] != "accepted" || record["handledThrough"] != "2026-07-12T10:30:00Z" {
|
||||
t.Fatalf("handled record = %#v", record)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBrowserInboxRejectsCaptureWithoutOpenVault(t *testing.T) {
|
||||
v := vault.NewVault(nil)
|
||||
app := &App{
|
||||
|
|
|
|||
Loading…
Reference in New Issue