package filemanagerrealtime import ( "encoding/json" "testing" "time" ) func TestHubSubscribePublishAndClose(t *testing.T) { hub := NewHub(4, nil) sub, err := hub.Subscribe("01961f5c-6471-7b81-9df5-2e2e1cf7e498") if err != nil { t.Fatalf("subscribe: %v", err) } if got := hub.SubscriberCount("01961f5c-6471-7b81-9df5-2e2e1cf7e498"); got != 1 { t.Fatalf("expected subscriber count 1, got %d", got) } delivered := hub.PublishToUsers([]string{"01961f5c-6471-7b81-9df5-2e2e1cf7e498"}, FileStatusEvent{ EventType: FileStatusChangedEventType, FileUUID: "01961f5c-6471-7b81-9df5-2e2e1cf7e499", Status: "processing", Name: "report.pdf", UpdatedAt: time.Now().UTC().Format(time.RFC3339), }) if delivered != 1 { t.Fatalf("expected one delivered event, got %d", delivered) } select { case event := <-sub.Events(): if event.Status != "processing" { t.Fatalf("expected processing status, got %q", event.Status) } case <-time.After(time.Second): t.Fatalf("timed out waiting for event") } sub.Close() if got := hub.SubscriberCount("01961f5c-6471-7b81-9df5-2e2e1cf7e498"); got != 0 { t.Fatalf("expected subscriber count 0 after close, got %d", got) } delivered = hub.PublishToUsers([]string{"01961f5c-6471-7b81-9df5-2e2e1cf7e498"}, FileStatusEvent{ EventType: FileStatusChangedEventType, FileUUID: "01961f5c-6471-7b81-9df5-2e2e1cf7e499", Status: "validated", Name: "report.pdf", UpdatedAt: time.Now().UTC().Format(time.RFC3339), }) if delivered != 0 { t.Fatalf("expected no delivered event after close, got %d", delivered) } } func TestFileStatusEventJSONPayload(t *testing.T) { reason := "file bucket mismatch" event := FileStatusEvent{ EventType: FileStatusChangedEventType, FileUUID: "01961f5c-6471-7b81-9df5-2e2e1cf7e499", Status: "failed", Name: "report.pdf", FailureReason: &reason, UpdatedAt: "2026-04-05T07:29:34Z", } raw, err := json.Marshal(event) if err != nil { t.Fatalf("marshal: %v", err) } var decoded map[string]any if err := json.Unmarshal(raw, &decoded); err != nil { t.Fatalf("unmarshal: %v", err) } if decoded["event_type"] != FileStatusChangedEventType { t.Fatalf("unexpected event_type: %#v", decoded["event_type"]) } if decoded["status"] != "failed" { t.Fatalf("unexpected status: %#v", decoded["status"]) } if decoded["failure_reason"] != reason { t.Fatalf("unexpected failure_reason: %#v", decoded["failure_reason"]) } } func TestHubReplaysRecentEventsOnSubscribe(t *testing.T) { hub := NewHub(2, nil) userID := "01961f5c-6471-7b81-9df5-2e2e1cf7e498" // Publish without active subscribers. if delivered := hub.PublishToUsers([]string{userID}, FileStatusEvent{ EventType: FileStatusChangedEventType, FileUUID: "01961f5c-6471-7b81-9df5-2e2e1cf7e499", Status: "processing", Name: "report.pdf", UpdatedAt: time.Now().UTC().Format(time.RFC3339), }); delivered != 0 { t.Fatalf("expected zero delivered before subscription, got %d", delivered) } sub, err := hub.Subscribe(userID) if err != nil { t.Fatalf("subscribe: %v", err) } defer sub.Close() select { case event := <-sub.Events(): if event.Status != "processing" { t.Fatalf("expected replayed processing status, got %q", event.Status) } case <-time.After(time.Second): t.Fatalf("timed out waiting for replayed event") } } func TestHubSubscribeWithFilterSkipsUnmatchedEvents(t *testing.T) { hub := NewHub(2, nil) userID := "01961f5c-6471-7b81-9df5-2e2e1cf7e498" keepFileID := "01961f5c-6471-7b81-9df5-2e2e1cf7e499" skipFileID := "01961f5c-6471-7b81-9df5-2e2e1cf7e500" sub, err := hub.SubscribeWithFilter(userID, func(event FileStatusEvent) bool { return event.FileUUID == keepFileID }) if err != nil { t.Fatalf("subscribe with filter: %v", err) } defer sub.Close() if delivered := hub.PublishToUsers([]string{userID}, FileStatusEvent{ EventType: FileStatusChangedEventType, FileUUID: skipFileID, Status: "processing", Name: "skip.pdf", UpdatedAt: time.Now().UTC().Format(time.RFC3339), }); delivered != 0 { t.Fatalf("expected unmatched event to be skipped, got %d delivered", delivered) } if delivered := hub.PublishToUsers([]string{userID}, FileStatusEvent{ EventType: FileStatusChangedEventType, FileUUID: keepFileID, Status: "ready", Name: "keep.pdf", UpdatedAt: time.Now().UTC().Format(time.RFC3339), }); delivered != 1 { t.Fatalf("expected matched event to be delivered, got %d", delivered) } select { case event := <-sub.Events(): if event.FileUUID != keepFileID { t.Fatalf("expected filtered event, got %q", event.FileUUID) } case <-time.After(time.Second): t.Fatalf("timed out waiting for filtered event") } select { case event := <-sub.Events(): t.Fatalf("unexpected extra event delivered: %#v", event) case <-time.After(50 * time.Millisecond): } }