ds-galo/backend/internal/app/album_live_test.go

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