fix(realtime): доставлять обложку альбома после готовности вариантов

Карточка альбома не получала ссылку на обложку: она появляется в объекте
альбома только после генерации thumb/medium, а событие об этом шаге в
персональные потоки не публиковалось — счётчик обновлялся, обложка нет.

- albums.Service.NotifyCoverReady публикует album.updated получателям, когда
  варианты готовы у фото, являющегося текущей обложкой (иначе no-op, чтобы
  массовая загрузка не рассылала событие на каждый файл)
- photos: порт AlbumAccess расширен, worker вызывает публикацию после
  photo.updated
- project.md §11/§12: триггер добавлен в контракт персонального потока
This commit is contained in:
iu
2026-10-16 12:47:50 +03:00
parent 742745e6aa
commit 44cb7c7d68
7 changed files with 129 additions and 11 deletions
+17
View File
@@ -80,6 +80,23 @@ func (s *Service) NotifyAlbumChanged(ctx context.Context, albumID string) {
s.PublishAlbumTo(ctx, albumID, EventAlbumUpdated, s.albumAudience(ctx, albumID))
}
// NotifyCoverReady republishes the album to personal streams when the photo that
// is currently its cover gains its variant URLs. The album object only carries a
// cover link once thumb/medium exist, so without this event an album card keeps
// showing a placeholder. Readiness of any other photo does not change the album
// object and is ignored, which keeps bulk uploads from fanning out an event per
// file.
func (s *Service) NotifyCoverReady(ctx context.Context, albumID, photoID string) {
a, err := s.store.GetByID(ctx, albumID)
if err != nil {
return
}
if a.CoverPhotoID == nil || *a.CoverPhotoID != photoID {
return
}
s.NotifyAlbumChanged(ctx, albumID)
}
// albumAudience resolves the users an album is visible to. Resolution failures
// are swallowed: a missed fan-out is recovered by the client's reconnect sync
// and must not fail the mutation that triggered it.
+51 -1
View File
@@ -314,7 +314,57 @@ func TestNotifyAlbumChangedReachesRecipientsOnly(t *testing.T) {
assertNoEvent(t, albumSub)
}
// photos_count counts only photos outside the trash.
// Cover readiness republishes the album only when the photo that became ready is
// the current cover; any other photo leaves the album object unchanged.
func TestNotifyCoverReadyPublishesOnlyForTheCover(t *testing.T) {
ctx := context.Background()
st := newStack(t)
owner := seedUser(t, st, "Owner")
member := seedUser(t, st, "Member")
album, err := st.svc.Create(ctx, owner, albums.CreateInput{Title: "Cover"})
if err != nil {
t.Fatalf("create: %v", err)
}
if err := st.repos.Members.Add(ctx, album.ID, member, "member", &owner); err != nil {
t.Fatalf("add member: %v", err)
}
const coverID = "photo-cover"
if _, err := st.db.Exec(
`INSERT INTO photos (id, album_id, uploader_id) VALUES ($1, $2, $3)`,
coverID, album.ID, owner,
); err != nil {
t.Fatalf("seed cover photo: %v", err)
}
if _, err := st.db.Exec(
`UPDATE albums SET cover_photo_id = $1 WHERE id = $2`, coverID, album.ID,
); err != nil {
t.Fatalf("set cover: %v", err)
}
ownerSub := st.hub.SubscribeUser(owner)
defer ownerSub.Close()
memberSub := st.hub.SubscribeUser(member)
defer memberSub.Close()
// A photo that is not the cover changes nothing.
st.svc.NotifyCoverReady(ctx, album.ID, "photo-other")
assertNoEvent(t, ownerSub)
assertNoEvent(t, memberSub)
// The cover becoming ready reaches the owner and every member.
st.svc.NotifyCoverReady(ctx, album.ID, coverID)
for name, sub := range map[string]*sse.Subscription{"owner": ownerSub, "member": memberSub} {
ev := nextEvent(t, sub)
if ev.Type != albums.EventAlbumUpdated {
t.Fatalf("%s event = %q, want %q", name, ev.Type, albums.EventAlbumUpdated)
}
if a, ok := ev.Data.(*models.Album); !ok || a.ID != album.ID {
t.Fatalf("%s payload = %#v", name, ev.Data)
}
}
}
func TestPhotosCountExcludesTrash(t *testing.T) {
ctx := context.Background()
st := newStack(t)
+38 -5
View File
@@ -7,6 +7,7 @@ import (
"image/color"
"image/png"
"os"
"sync"
"testing"
"time"
@@ -52,6 +53,35 @@ func (stubAlbums) Access(ctx context.Context, albumID, userID string) (*models.A
func (stubAlbums) NotifyAlbumChanged(context.Context, string) {}
func (stubAlbums) NotifyCoverReady(context.Context, string, string) {}
// recordingAlbums records cover-readiness announcements so worker tests can
// assert the worker reports them.
type recordingAlbums struct {
stubAlbums
mu sync.Mutex
noticed map[string]struct{}
}
func newRecordingAlbums() *recordingAlbums {
return &recordingAlbums{noticed: map[string]struct{}{}}
}
func (r *recordingAlbums) NotifyCoverReady(_ context.Context, albumID, photoID string) {
r.mu.Lock()
defer r.mu.Unlock()
r.noticed[albumID+"|"+photoID] = struct{}{}
}
// coverReady reports whether the given album/photo pair was announced.
func (r *recordingAlbums) coverReady(albumID, photoID string) bool {
r.mu.Lock()
defer r.mu.Unlock()
_, ok := r.noticed[albumID+"|"+photoID]
return ok
}
func createUser(t *testing.T, ctx context.Context, pool *pgxpool.Pool) string {
t.Helper()
id := auth.NewID()
@@ -126,19 +156,22 @@ func newTestService(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend
return NewService(NewStore(pool), stubAlbums{}, cl, sse.NewHub(1)), cl
}
// newTestServiceFS builds a service backed by a temporary local file-system
// storage backend so the full flow can run without external infrastructure.
func newTestServiceFS(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend) {
// newRecordingServiceFS builds a service backed by a temporary local file-system
// storage backend (so the full flow runs without external infrastructure) with an
// album port that records cover-readiness announcements, for the worker's
// publication assertions.
func newRecordingServiceFS(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend, *recordingAlbums) {
t.Helper()
signer, err := storage.NewSigner("test-secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("storage init: %v", err)
t.Fatalf("signer init: %v", err)
}
fs, err := storage.NewFS(t.TempDir(), signer)
if err != nil {
t.Fatalf("storage init: %v", err)
}
return NewService(NewStore(pool), stubAlbums{}, fs, sse.NewHub(1)), fs
rec := newRecordingAlbums()
return NewService(NewStore(pool), rec, fs, sse.NewHub(1)), fs, rec
}
// cleanupPhoto removes a photo, its objects (rows and blobs) and bindings.
+3
View File
@@ -29,6 +29,9 @@ type AlbumAccess interface {
// NotifyAlbumChanged announces a change to the album's photo composition on
// the recipients' personal streams (album cards and counts update live).
NotifyAlbumChanged(ctx context.Context, albumID string)
// NotifyCoverReady announces the album when the photo that is its cover gains
// variant URLs: only then does the album object carry a cover link.
NotifyCoverReady(ctx context.Context, albumID, photoID string)
}
type Service struct {
+3
View File
@@ -454,6 +454,9 @@ func (w *Worker) notify(ctx context.Context, photoID string) {
// Variant readiness updates an existing photo; it is not a new photo in the
// album, so clients update it in place instead of inserting at the top.
w.svc.hub.Publish(p.AlbumID, "photo.updated", p)
// The album object only carries a cover link once its variants exist, so the
// cover's readiness is what actually refreshes album cards.
w.svc.albums.NotifyCoverReady(ctx, p.AlbumID, p.ID)
}
func (w *Worker) fail(ctx context.Context, photoID, msg string) {
+10 -4
View File
@@ -15,16 +15,16 @@ import (
func TestWorkerGeneratesVariantsAndCollects(t *testing.T) {
pool := testPool(t)
svc, cl := newTestService(t, pool)
runWorkerFlow(t, pool, svc, cl)
runWorkerFlow(t, pool, svc, cl, nil)
}
func TestWorkerGeneratesVariantsWithFS(t *testing.T) {
pool := testPool(t)
svc, cl := newTestServiceFS(t, pool)
runWorkerFlow(t, pool, svc, cl)
svc, cl, albums := newRecordingServiceFS(t, pool)
runWorkerFlow(t, pool, svc, cl, albums)
}
func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Backend) {
func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Backend, albums *recordingAlbums) {
t.Helper()
ctx := context.Background()
user := createUser(t, ctx, pool)
@@ -57,6 +57,12 @@ func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Ba
t.Fatalf("no variant-ready event published")
}
// Variant readiness must also announce the album to personal streams, because
// the album object only gains a cover link once the variants exist.
if albums != nil && !albums.coverReady(album, p.ID) {
t.Fatalf("cover readiness was not announced for the uploaded photo")
}
refreshed, err := svc.store.GetByID(ctx, p.ID, "")
if err != nil {
t.Fatalf("get refreshed: %v", err)
+7 -1
View File
@@ -464,7 +464,7 @@ data: {"id":"...","album_id":"...","thumb_url":"...","uploader":{...}}
|---|---|---|
| `notification` | создано уведомление для пользователя | объект уведомления |
| `album.created` | альбом появился в списке пользователя (он его создал или принял инвайт) | объект альбома |
| `album.updated` | изменены название/описание/обложка, состав участников или состав фото | объект альбома |
| `album.updated` | изменены название/описание/обложка, состав участников, состав фото или готовы варианты фото-обложки | объект альбома |
| `album.deleted` | альбом удалён владельцем или доступ отозван (исключение из участников) | `{"id":"..."}` |
Объект альбома в `album.created` и `album.updated` совпадает по полям с элементом
@@ -481,6 +481,12 @@ data: {"id":"...","album_id":"...","thumb_url":"...","uploader":{...}}
страница узнаёт об изменениях из `photo.*`). `photos_count` считает только фото вне
корзины и совпадает с числом фото в сетке альбома.
Ссылка на обложку (`cover_thumb_url`, `cover_medium_url`) появляется в объекте альбома
только после генерации вариантов thumb/medium, поэтому готовность вариантов у фото,
являющегося текущей обложкой, тоже публикует `album.updated` получателям — иначе карточки
альбомов остаются с заглушкой вместо обложки. Готовность вариантов прочих фото объект
альбома не меняет и события не порождает.
### Backend: hub на каналах
- В памяти `map[albumID]map[clientID]chan Event`
- Регистрация клиента при подключении, удаление при отключении