feat(realtime): доставлять клиентам события жизненного цикла альбома и изменения состава фото

Публиковать album.created/updated/deleted и изменения состава фото в личный поток SSE, а также photo.created при загрузке, чтобы списки альбомов, счётчики и открытые страницы альбома обновлялись без перезагрузки. - members.Repository получил Audience (владелец + участники) для дешёвого веера - albums.Service рассылает события в личные потоки получателей, перечитывая альбом через GetByID и attachCover, чтобы payload совпадал с GET /api/albums - members.Service публикует при принятии инвайта, смене роли и исключении - photos.Service публикует photo.created при загрузке и album.updated при добавлении/копировании/перемещении/восстановлении/корзине/окончательном удалении - photos_count теперь считает фото вне корзины - web-клиент подписывается на /api/sse/me один раз на уровне layout'а (app) и применяет события к стору альбомов; страница альбома уходит при потере доступа
2026-10-15 23:25:07 +03:00
parent cbff7a340c
commit e82f45647c
31 changed files with 2157 additions and 28 deletions
@@ -0,0 +1,457 @@
+package app
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"io/fs"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+
+	assets "galo"
+	"galo/internal/auth"
+	"galo/internal/config"
+	"galo/internal/db"
+	"galo/internal/sse"
+	"galo/internal/storage"
+	"galo/internal/store"
+)
+
+// newAlbumLiveServer starts the real router over a migrated SQLite database and
+// a filesystem blob store, so personal-stream behaviour is exercised through
+// HTTP exactly as clients see it.
+func newAlbumLiveServer(t *testing.T) *httptest.Server {
+	t.Helper()
+	gin.SetMode(gin.TestMode)
+
+	path := filepath.Join(t.TempDir(), "meta.db")
+	migFS, err := fs.Sub(assets.Migrations, "migrations/sqlite")
+	if err != nil {
+		t.Fatalf("migrations fs: %v", err)
+	}
+	if err := db.MigrateSQLite(path, migFS); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	database, err := db.OpenSQLite(path)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	t.Cleanup(func() { _ = database.Close() })
+
+	blob, err := storage.NewFS(t.TempDir(), mustSigner(t))
+	if err != nil {
+		t.Fatalf("storage: %v", err)
+	}
+	tm := auth.NewTokenManager("album-live-secret", 15*time.Minute, time.Hour)
+	srv := New(&config.Config{}, store.NewSQLite(database), database.PingContext, tm, blob, sse.NewHub(128))
+
+	ts := httptest.NewServer(srv.Router())
+	t.Cleanup(ts.Close)
+	return ts
+}
+
+func doJSON(t *testing.T, method, url, token string, body any) (int, map[string]any) {
+	t.Helper()
+	var buf bytes.Buffer
+	if body != nil {
+		if err := json.NewEncoder(&buf).Encode(body); err != nil {
+			t.Fatalf("encode body: %v", err)
+		}
+	}
+	req, err := http.NewRequest(method, url, &buf)
+	if err != nil {
+		t.Fatalf("request: %v", err)
+	}
+	if body != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("%s %s: %v", method, url, err)
+	}
+	defer resp.Body.Close()
+	raw, _ := io.ReadAll(resp.Body)
+	out := map[string]any{}
+	if len(raw) > 0 {
+		_ = json.Unmarshal(raw, &out)
+	}
+	return resp.StatusCode, out
+}
+
+func registerUser(t *testing.T, base, email string) (token, id string) {
+	t.Helper()
+	status, body := doJSON(t, http.MethodPost, base+"/api/auth/register", "", map[string]any{
+		"email": email, "password": "password123", "name": email,
+	})
+	if status != http.StatusCreated {
+		t.Fatalf("register %s: status %d body %v", email, status, body)
+	}
+	return body["tokens"].(map[string]any)["access_token"].(string),
+		body["user"].(map[string]any)["id"].(string)
+}
+
+func createAlbum(t *testing.T, base, token, title string) string {
+	t.Helper()
+	status, body := doJSON(t, http.MethodPost, base+"/api/albums", token, map[string]any{"title": title})
+	if status != http.StatusCreated {
+		t.Fatalf("create album: status %d body %v", status, body)
+	}
+	return body["album"].(map[string]any)["id"].(string)
+}
+
+type streamEvent struct {
+	typ  string
+	data string
+}
+
+// albumPayload is the subset of the album object the assertions inspect.
+type albumPayload struct {
+	ID           string `json:"id"`
+	Title        string `json:"title"`
+	MembersCount int    `json:"members_count"`
+	PhotosCount  int    `json:"photos_count"`
+}
+
+func decodeAlbum(t *testing.T, data string) albumPayload {
+	t.Helper()
+	var a albumPayload
+	if err := json.Unmarshal([]byte(data), &a); err != nil {
+		t.Fatalf("decode album %q: %v", data, err)
+	}
+	return a
+}
+
+type personalStream struct {
+	events chan streamEvent
+	cancel context.CancelFunc
+}
+
+func openPersonalStream(t *testing.T, base, token string) *personalStream {
+	t.Helper()
+	return openStream(t, base, "/api/sse/me?token="+token)
+}
+
+func openAlbumStream(t *testing.T, base, albumID, token string) *personalStream {
+	t.Helper()
+	return openStream(t, base, "/api/sse/albums/"+albumID+"?token="+token)
+}
+
+func openStream(t *testing.T, base, path string) *personalStream {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+path, nil)
+	if err != nil {
+		cancel()
+		t.Fatalf("sse request: %v", err)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		cancel()
+		t.Fatalf("sse connect: %v", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		resp.Body.Close()
+		cancel()
+		t.Fatalf("sse status = %d, want 200", resp.StatusCode)
+	}
+
+	s := &personalStream{events: make(chan streamEvent, 128), cancel: cancel}
+	go func() {
+		defer resp.Body.Close()
+		sc := bufio.NewScanner(resp.Body)
+		sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+		var typ string
+		for sc.Scan() {
+			line := sc.Text()
+			switch {
+			case strings.HasPrefix(line, "event: "):
+				typ = strings.TrimPrefix(line, "event: ")
+			case strings.HasPrefix(line, "data: "):
+				select {
+				case s.events <- streamEvent{typ, strings.TrimPrefix(line, "data: ")}:
+				default:
+				}
+			}
+		}
+	}()
+	t.Cleanup(cancel)
+	return s
+}
+
+func (s *personalStream) next(t *testing.T) streamEvent {
+	t.Helper()
+	select {
+	case ev := <-s.events:
+		return ev
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for a personal event")
+		return streamEvent{}
+	}
+}
+
+func (s *personalStream) expectNoEvent(t *testing.T) {
+	t.Helper()
+	select {
+	case ev := <-s.events:
+		t.Fatalf("unexpected event %q", ev.typ)
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+// TestAlbumLifecycleReachesPersonalStreams drives the whole album lifecycle over
+// HTTP for an owner and a member and asserts what each personal stream receives.
+func TestAlbumLifecycleReachesPersonalStreams(t *testing.T) {
+	ts := newAlbumLiveServer(t)
+	base := ts.URL
+
+	ownerTok, _ := registerUser(t, base, "owner@live.local")
+	memberTok, memberID := registerUser(t, base, "member@live.local")
+
+	ownerSSE := openPersonalStream(t, base, ownerTok)
+	memberSSE := openPersonalStream(t, base, memberTok)
+
+	// Creating an album tells the owner.
+	albumID := createAlbum(t, base, ownerTok, "Trip")
+	if ev := ownerSSE.next(t); ev.typ != "album.created" {
+		t.Fatalf("owner event = %q, want album.created", ev.typ)
+	} else if a := decodeAlbum(t, ev.data); a.ID != albumID || a.MembersCount != 0 {
+		t.Fatalf("album.created payload = %+v", a)
+	}
+
+	// Inviting and accepting gives the invitee the album and updates the owner.
+	status, invBody := doJSON(t, http.MethodPost, base+"/api/albums/"+albumID+"/invites", ownerTok,
+		map[string]any{"role": "member", "max_uses": 1, "expires_in": "1h"})
+	if status != http.StatusCreated {
+		t.Fatalf("create invite: status %d body %v", status, invBody)
+	}
+	inviteToken := invBody["invite"].(map[string]any)["token"].(string)
+	if status, body := doJSON(t, http.MethodPost, base+"/api/invites/"+inviteToken+"/accept", memberTok, nil); status != http.StatusOK {
+		t.Fatalf("accept invite: status %d body %v", status, body)
+	}
+
+	if ev := memberSSE.next(t); ev.typ != "album.created" {
+		t.Fatalf("invitee event = %q, want album.created", ev.typ)
+	} else if a := decodeAlbum(t, ev.data); a.ID != albumID || a.MembersCount != 1 {
+		t.Fatalf("invitee payload = %+v", a)
+	}
+	if ev := ownerSSE.next(t); ev.typ != "album.updated" {
+		t.Fatalf("owner event = %q, want album.updated", ev.typ)
+	} else if a := decodeAlbum(t, ev.data); a.MembersCount != 1 {
+		t.Fatalf("owner payload = %+v", a)
+	}
+	// The owner also receives the persistent notification for the new member.
+	if ev := ownerSSE.next(t); ev.typ != "notification" {
+		t.Fatalf("owner event = %q, want notification", ev.typ)
+	}
+
+	// Changing the member's role reaches both.
+	if status, body := doJSON(t, http.MethodPatch, base+"/api/albums/"+albumID+"/members/"+memberID, ownerTok,
+		map[string]any{"role": "editor"}); status != http.StatusNoContent {
+		t.Fatalf("change role: status %d body %v", status, body)
+	}
+	for _, s := range []*personalStream{ownerSSE, memberSSE} {
+		if ev := s.next(t); ev.typ != "album.updated" {
+			t.Fatalf("role change event = %q, want album.updated", ev.typ)
+		}
+	}
+
+	// Removing the member updates the owner and tells the removed member the
+	// album is gone, carrying only an id.
+	if status, body := doJSON(t, http.MethodDelete, base+"/api/albums/"+albumID+"/members/"+memberID, ownerTok, nil); status != http.StatusNoContent {
+		t.Fatalf("remove member: status %d body %v", status, body)
+	}
+	if ev := ownerSSE.next(t); ev.typ != "album.updated" {
+		t.Fatalf("owner event = %q, want album.updated", ev.typ)
+	} else if a := decodeAlbum(t, ev.data); a.MembersCount != 0 {
+		t.Fatalf("owner payload = %+v", a)
+	}
+	if ev := memberSSE.next(t); ev.typ != "album.deleted" {
+		t.Fatalf("removed member event = %q, want album.deleted", ev.typ)
+	} else {
+		var raw map[string]any
+		if err := json.Unmarshal([]byte(ev.data), &raw); err != nil {
+			t.Fatalf("decode deleted payload: %v", err)
+		}
+		if raw["id"] != albumID || raw["owner_id"] != nil {
+			t.Fatalf("album.deleted payload = %v", raw)
+		}
+	}
+	memberSSE.expectNoEvent(t)
+
+	// Deleting the album reaches the owner, and the list reconciles without it.
+	if status, body := doJSON(t, http.MethodDelete, base+"/api/albums/"+albumID, ownerTok, nil); status != http.StatusNoContent {
+		t.Fatalf("delete album: status %d body %v", status, body)
+	}
+	if ev := ownerSSE.next(t); ev.typ != "album.deleted" {
+		t.Fatalf("owner event = %q, want album.deleted", ev.typ)
+	}
+
+	status, listBody := doJSON(t, http.MethodGet, base+"/api/albums", memberTok, nil)
+	if status != http.StatusOK {
+		t.Fatalf("list albums: status %d", status)
+	}
+	if albums := listBody["albums"].([]any); len(albums) != 0 {
+		t.Fatalf("albums after delete = %d, want 0", len(albums))
+	}
+}
+
+func uploadPhoto(t *testing.T, base, token, albumID, filename string) string {
+	t.Helper()
+	body, contentType := multipartFile(t, "file", filename, makeTestPNG(t))
+	req, err := http.NewRequest(http.MethodPost, base+"/api/albums/"+albumID+"/photos", body)
+	if err != nil {
+		t.Fatalf("upload request: %v", err)
+	}
+	req.Header.Set("Content-Type", contentType)
+	req.Header.Set("Authorization", "Bearer "+token)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("upload: %v", err)
+	}
+	defer resp.Body.Close()
+	raw, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusCreated {
+		t.Fatalf("upload status = %d body %s", resp.StatusCode, raw)
+	}
+	var out struct {
+		Photos []struct {
+			ID string `json:"id"`
+		} `json:"photos"`
+	}
+	if err := json.Unmarshal(raw, &out); err != nil || len(out.Photos) == 0 {
+		t.Fatalf("upload decode: %v body %s", err, raw)
+	}
+	return out.Photos[0].ID
+}
+
+// TestPhotoChangesUpdateAlbumCountsAndStreams covers photo composition changes:
+// upload publishes photo.created on the album stream and album.updated with the
+// new count to the owner and members; trashing and restoring move the count back
+// and forth (the counter ignores photos in the trash).
+func TestPhotoChangesUpdateAlbumCountsAndStreams(t *testing.T) {
+	ts := newAlbumLiveServer(t)
+	base := ts.URL
+
+	ownerTok, _ := registerUser(t, base, "photo-owner@live.local")
+	memberTok, _ := registerUser(t, base, "photo-member@live.local")
+
+	albumID := createAlbum(t, base, ownerTok, "Photos")
+	status, invBody := doJSON(t, http.MethodPost, base+"/api/albums/"+albumID+"/invites", ownerTok,
+		map[string]any{"role": "member", "max_uses": 1, "expires_in": "1h"})
+	if status != http.StatusCreated {
+		t.Fatalf("create invite: status %d body %v", status, invBody)
+	}
+	inviteToken := invBody["invite"].(map[string]any)["token"].(string)
+	if status, body := doJSON(t, http.MethodPost, base+"/api/invites/"+inviteToken+"/accept", memberTok, nil); status != http.StatusOK {
+		t.Fatalf("accept invite: status %d body %v", status, body)
+	}
+
+	// Streams open after the membership is settled, so only photo changes flow.
+	ownerSSE := openPersonalStream(t, base, ownerTok)
+	memberSSE := openPersonalStream(t, base, memberTok)
+	albumStream := openAlbumStream(t, base, albumID, ownerTok)
+
+	awaitAlbum := func(context string, wantCount int) {
+		t.Helper()
+		for name, s := range map[string]*personalStream{"owner": ownerSSE, "member": memberSSE} {
+			ev := s.next(t)
+			if ev.typ != "album.updated" {
+				t.Fatalf("%s/%s: event = %q, want album.updated", context, name, ev.typ)
+			}
+			if a := decodeAlbum(t, ev.data); a.PhotosCount != wantCount {
+				t.Fatalf("%s/%s: photos_count = %d, want %d", context, name, a.PhotosCount, wantCount)
+			}
+		}
+	}
+
+	// Upload: the photo appears in the album and the count grows.
+	photoID := uploadPhoto(t, base, ownerTok, albumID, "a.png")
+	if ev := albumStream.next(t); ev.typ != "photo.created" {
+		t.Fatalf("album stream event = %q, want photo.created", ev.typ)
+	}
+	awaitAlbum("upload", 1)
+
+	// Trash: the photo leaves the album and the count drops.
+	if status, body := doJSON(t, http.MethodDelete, base+"/api/photos/"+photoID, ownerTok, nil); status != http.StatusNoContent {
+		t.Fatalf("trash photo: status %d body %v", status, body)
+	}
+	if ev := albumStream.next(t); ev.typ != "photo.deleted" {
+		t.Fatalf("album stream event = %q, want photo.deleted", ev.typ)
+	}
+	awaitAlbum("trash", 0)
+
+	// Restore: the photo is back and the count grows again.
+	if status, body := doJSON(t, http.MethodPost, base+"/api/photos/"+photoID+"/restore", ownerTok, nil); status != http.StatusOK {
+		t.Fatalf("restore photo: status %d body %v", status, body)
+	}
+	if ev := albumStream.next(t); ev.typ != "photo.created" {
+		t.Fatalf("album stream event = %q, want photo.created", ev.typ)
+	}
+	awaitAlbum("restore", 1)
+}
+
+// collectAlbumUpdates reads n album.updated events and maps album id -> count.
+func collectAlbumUpdates(t *testing.T, s *personalStream, n int) map[string]int {
+	t.Helper()
+	out := map[string]int{}
+	for i := 0; i < n; i++ {
+		ev := s.next(t)
+		if ev.typ != "album.updated" {
+			t.Fatalf("event = %q, want album.updated", ev.typ)
+		}
+		a := decodeAlbum(t, ev.data)
+		out[a.ID] = a.PhotosCount
+	}
+	return out
+}
+
+// Moving a photo updates the counts of both albums; copying updates the target.
+func TestPhotoMoveAndCopyUpdateAlbumCounts(t *testing.T) {
+	ts := newAlbumLiveServer(t)
+	base := ts.URL
+
+	ownerTok, _ := registerUser(t, base, "mover@live.local")
+	personal := openPersonalStream(t, base, ownerTok)
+
+	albumA := createAlbum(t, base, ownerTok, "A")
+	albumB := createAlbum(t, base, ownerTok, "B")
+	for i := 0; i < 2; i++ {
+		if ev := personal.next(t); ev.typ != "album.created" {
+			t.Fatalf("event = %q, want album.created", ev.typ)
+		}
+	}
+
+	photoID := uploadPhoto(t, base, ownerTok, albumA, "a.png")
+	if got := collectAlbumUpdates(t, personal, 1); got[albumA] != 1 {
+		t.Fatalf("after upload A = %d, want 1 (%v)", got[albumA], got)
+	}
+
+	status, body := doJSON(t, http.MethodPost, base+"/api/photo-batch/move", ownerTok,
+		map[string]any{"ids": []string{photoID}, "album_id": albumB})
+	if status != http.StatusOK {
+		t.Fatalf("move: status %d body %v", status, body)
+	}
+	if got := collectAlbumUpdates(t, personal, 2); got[albumA] != 0 || got[albumB] != 1 {
+		t.Fatalf("after move A=%d B=%d, want 0/1 (%v)", got[albumA], got[albumB], got)
+	}
+
+	status, body = doJSON(t, http.MethodPost, base+"/api/photo-batch/copy", ownerTok,
+		map[string]any{"ids": []string{photoID}, "album_id": albumA})
+	if status != http.StatusOK {
+		t.Fatalf("copy: status %d body %v", status, body)
+	}
+	if got := collectAlbumUpdates(t, personal, 1); got[albumA] != 1 {
+		t.Fatalf("after copy A = %d, want 1 (%v)", got[albumA], got)
+	}
+}