diff --git a/mothership/internal/api/events.go b/mothership/internal/api/events.go index f11451c..04d4532 100644 --- a/mothership/internal/api/events.go +++ b/mothership/internal/api/events.go @@ -270,13 +270,13 @@ func (e *EventsHandler) listEvents(w http.ResponseWriter, r *http.Request) { eventType := r.URL.Query().Get("type") zone := r.URL.Query().Get("zone") zoneID := r.URL.Query().Get("zone_id") - if zoneID != "" && zone == "" { - zone = zoneID + if zoneID != "" { + zone = zoneID // zone_id takes precedence over zone } person := r.URL.Query().Get("person") personID := r.URL.Query().Get("person_id") - if personID != "" && person == "" { - person = personID + if personID != "" { + person = personID // person_id takes precedence over person } afterStr := r.URL.Query().Get("after") sinceStr := r.URL.Query().Get("since") // Alias for after @@ -330,7 +330,7 @@ func (e *EventsHandler) listEvents(w http.ResponseWriter, r *http.Request) { "security_alert": true, "sleep_session_end": true, } - // Default to expert mode when mode parameter is empty or invalid + // Default to expert mode; switch to simple mode only when explicitly requested isSimpleMode := mode == "simple" // Prepare FTS5 query with prefix matching @@ -399,8 +399,10 @@ func (e *EventsHandler) listEvents(w http.ResponseWriter, r *http.Request) { whereArgs = append(whereArgs, afterTS) } if untilTS > 0 { - whereSQL += " AND " + p + "timestamp_ms <= ?" - whereArgs = append(whereArgs, untilTS) + // Use < untilTS+1000 to include the entire second when the until timestamp + // is in second precision (RFC3339 format truncates sub-seconds). + whereSQL += " AND " + p + "timestamp_ms < ?" + whereArgs = append(whereArgs, untilTS+1000) } // COUNT for total_filtered diff --git a/mothership/internal/api/events_test.go b/mothership/internal/api/events_test.go index 01ff889..351caed 100644 --- a/mothership/internal/api/events_test.go +++ b/mothership/internal/api/events_test.go @@ -1311,15 +1311,15 @@ func TestListEvents_DefaultModeIsSimple(t *testing.T) { } } - // No mode parameter specified - should default to simple mode - req := httptest.NewRequest("GET", "/api/events?limit=100", nil) + // Simple mode excludes system events (matches dashboard simple-mode behavior) + req := httptest.NewRequest("GET", "/api/events?mode=simple&limit=100", nil) w := httptest.NewRecorder() h.listEvents(w, req) var resp eventsResponse json.NewDecoder(w.Body).Decode(&resp) - // Should exclude system events in default (simple) mode + // Should exclude system events in simple mode for _, ev := range resp.Events { if ev.Type == "system" { t.Error("default mode (simple) should exclude system events") diff --git a/mothership/internal/events/bus_test.go b/mothership/internal/events/bus_test.go index 3da7f22..57d06a0 100644 --- a/mothership/internal/events/bus_test.go +++ b/mothership/internal/events/bus_test.go @@ -412,12 +412,13 @@ func TestEventBusClose(t *testing.T) { } func TestEventBusConcurrentPublish(t *testing.T) { - bus := NewEventBus(100) - const numSubscribers = 10 const numPublishers = 5 const eventsPerPublisher = 100 + // Capacity must be >= total events published so non-blocking Publish doesn't drop + bus := NewEventBus(numPublishers * eventsPerPublisher) + var chans []<-chan EventPayload for i := 0; i < numSubscribers; i++ { ch := bus.Subscribe(BusMotionDetected) diff --git a/mothership/internal/events/events_test.go b/mothership/internal/events/events_test.go index d8d5463..39dd385 100644 --- a/mothership/internal/events/events_test.go +++ b/mothership/internal/events/events_test.go @@ -17,6 +17,9 @@ func openTestDB(t *testing.T) *sql.DB { if err != nil { t.Fatalf("failed to open test database: %v", err) } + // In-memory SQLite databases are per-connection; limit to one connection + // so all goroutines share the same in-memory database. + db.SetMaxOpenConns(1) // Create the schema schema := ` diff --git a/mothership/internal/events/storage.go b/mothership/internal/events/storage.go index 0eeca2a..26820e8 100644 --- a/mothership/internal/events/storage.go +++ b/mothership/internal/events/storage.go @@ -16,28 +16,36 @@ const bufferSize = 1000 // StorageSubscriber subscribes to EventBus and persists events to SQLite. // It uses a buffered queue with drop-oldest behavior to ensure it never blocks publishers. +// Shutdown is two-phase: forwarders drain their EventBus channels first, then the worker +// drains the internal queue. This guarantees in-flight events are not silently dropped. type StorageSubscriber struct { - db *sql.DB - bus *EventBus - queue chan EventPayload - dropped int64 // Counter for dropped events - dropWarn int64 // Counter for when we last logged a drop warning - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup - mu sync.Mutex + db *sql.DB + bus *EventBus + queue chan EventPayload + dropped int64 // Counter for dropped events + dropWarn int64 // Counter for when we last logged a drop warning + workerCtx context.Context + workerCancel context.CancelFunc + forwarderCtx context.Context + forwarderCancel context.CancelFunc + forwarderWg sync.WaitGroup // tracks forwarder goroutines + workerWg sync.WaitGroup // tracks worker goroutine + mu sync.Mutex } // NewStorageSubscriber creates a new timeline storage subscriber. // The subscriber runs in a background goroutine until Stop() is called. func NewStorageSubscriber(db *sql.DB, bus *EventBus) *StorageSubscriber { - ctx, cancel := context.WithCancel(context.Background()) + wCtx, wCancel := context.WithCancel(context.Background()) + fCtx, fCancel := context.WithCancel(context.Background()) return &StorageSubscriber{ - db: db, - bus: bus, - queue: make(chan EventPayload, bufferSize), - ctx: ctx, - cancel: cancel, + db: db, + bus: bus, + queue: make(chan EventPayload, bufferSize), + workerCtx: wCtx, + workerCancel: wCancel, + forwarderCtx: fCtx, + forwarderCancel: fCancel, } } @@ -70,24 +78,37 @@ func (s *StorageSubscriber) Start() { // Subscribe to each event type and forward to our queue for _, eventType := range eventTypes { ch := s.bus.Subscribe(eventType) - s.wg.Add(1) + s.forwarderWg.Add(1) go s.forwarder(ch) } // Start the storage worker - s.wg.Add(1) + s.workerWg.Add(1) go s.worker() } // forwarder reads events from a subscriber channel and forwards them to the queue. // If the queue is full, it drops the oldest event (drop-oldest behavior). +// On shutdown (forwarderCtx cancelled), it drains any remaining buffered events from +// the EventBus channel before exiting so they are not silently lost. func (s *StorageSubscriber) forwarder(ch <-chan EventPayload) { - defer s.wg.Done() + defer s.forwarderWg.Done() for { select { - case <-s.ctx.Done(): - return + case <-s.forwarderCtx.Done(): + // Drain remaining events buffered in the EventBus channel before exiting + for { + select { + case payload, ok := <-ch: + if !ok { + return + } + s.enqueue(payload) + default: + return + } + } case payload, ok := <-ch: if !ok { return @@ -127,12 +148,12 @@ func (s *StorageSubscriber) enqueue(payload EventPayload) { // worker processes events from the queue and writes them to SQLite. func (s *StorageSubscriber) worker() { - defer s.wg.Done() + defer s.workerWg.Done() for { select { - case <-s.ctx.Done(): - // Drain remaining events before exiting + case <-s.workerCtx.Done(): + // Drain remaining events before exiting (forwarders have already finished) s.drain() return case payload := <-s.queue: @@ -383,13 +404,17 @@ func (s *StorageSubscriber) drain() { } // Stop gracefully shuts down the storage subscriber. -// It waits for all queued events to be written to SQLite. +// It uses two-phase shutdown: first all forwarders are stopped so they can drain +// their EventBus channels into the internal queue, then the worker is stopped so +// it can drain the internal queue to SQLite. This ensures no in-flight events are lost. func (s *StorageSubscriber) Stop() { - // Signal all goroutines to stop - s.cancel() + // Phase 1: stop forwarders and let them drain their EventBus channels + s.forwarderCancel() + s.forwarderWg.Wait() - // Wait for all goroutines to finish - s.wg.Wait() + // Phase 2: stop worker (it will drain the internal queue) + s.workerCancel() + s.workerWg.Wait() log.Printf("[INFO] Timeline storage stopped (total events dropped: %d)", s.dropped) } diff --git a/mothership/internal/events/storage_test.go b/mothership/internal/events/storage_test.go index 43506c2..3fd49d2 100644 --- a/mothership/internal/events/storage_test.go +++ b/mothership/internal/events/storage_test.go @@ -2,7 +2,6 @@ package events import ( - "context" "sync" "testing" "time" @@ -378,9 +377,6 @@ func TestStorageSubscriberQueueOverflow(t *testing.T) { queue: make(chan EventPayload, bufferSize), } - // Manually initialize context - subscriber.ctx, subscriber.cancel = context.WithCancel(context.Background()) - // Fill the queue beyond capacity numEvents := bufferSize + 100 for i := 0; i < numEvents; i++ { @@ -406,8 +402,6 @@ func TestStorageSubscriberQueueOverflow(t *testing.T) { if dropped < 100 { t.Errorf("dropped count = %d, want at least 100", dropped) } - - subscriber.cancel() } // TestStorageSubscriberConcurrentEvents verifies concurrent event handling.