fm_be/internal/realtime/filemanager/status_hub_test.go

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):
	}
}