From baa78588303e1067b9a943959d28d123c56ab814 Mon Sep 17 00:00:00 2001 From: itzjb Date: Mon, 10 Aug 2026 19:50:30 +0900 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20=EC=86=A1=EC=B6=9C=20=ED=86=A0?= =?UTF-8?q?=ED=81=B0=20=EB=AC=B4=ED=9A=A8=EB=A5=BC=20=EC=9E=AC=EC=97=B0?= =?UTF-8?q?=EA=B2=B0=20=ED=95=84=EC=9A=94=20=EC=83=81=ED=83=9C=EB=A1=9C=20?= =?UTF-8?q?=EA=B5=AC=EB=B6=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/auth/streaming_account.go | 29 ++++++++ .../auth/streaming_account_postgres_test.go | 25 +++++++ internal/auth/youtube_oauth.go | 11 +++ internal/auth/youtube_oauth_test.go | 68 +++++++++++++++++++ ..._add_streaming_reconnect_required.down.sql | 6 ++ ...05_add_streaming_reconnect_required.up.sql | 6 ++ internal/server/server.go | 4 ++ internal/server/stream_test.go | 7 ++ internal/streaming/youtube_test.go | 12 ++++ 9 files changed, 168 insertions(+) create mode 100644 internal/database/migration/sql/000005_add_streaming_reconnect_required.down.sql create mode 100644 internal/database/migration/sql/000005_add_streaming_reconnect_required.up.sql diff --git a/internal/auth/streaming_account.go b/internal/auth/streaming_account.go index f56a209..db56043 100644 --- a/internal/auth/streaming_account.go +++ b/internal/auth/streaming_account.go @@ -47,6 +47,12 @@ type StreamingAccount struct { // NULL이다. 이 값을 추적하지 않으면 연결이 만료 후 조용히 죽는다. RefreshTokenExpiresAt *time.Time + // 토큰 갱신이 "무효 토큰"으로 거절된 시각. 사용자가 플랫폼 쪽에서 권한을 + // 취소하는 등 재연결 없이는 복구되지 않는 상태의 표식이며, 재연결(Upsert) + // 시 NULL로 리셋된다. 조회 API가 "재연결 필요"를 API 호출 없이 판별하는 + // 근거다. + ReconnectRequiredAt *time.Time + // 치지직처럼 ingest URL을 API로 제공하지 않는 플랫폼을 위한 수동 설정값. // 연결(OAuth) 플로우는 이 컬럼을 건드리지 않는다. ManualIngestURL *string `gorm:"type:text"` @@ -95,6 +101,8 @@ type StreamingAccountStore interface { UpdateRefreshToken(ctx context.Context, id uuid.UUID, ciphertext []byte, version *int16, expiresAt *time.Time) error // UpdateStreamInfo는 프리로딩된 재사용 스트림 정보를 행 락 하에 저장한다. UpdateStreamInfo(ctx context.Context, id uuid.UUID, info StreamInfo) error + // MarkReconnectRequired는 토큰 갱신이 무효 토큰으로 거절됐음을 기록한다. + MarkReconnectRequired(ctx context.Context, id uuid.UUID, at time.Time) error } type gormStreamingAccountStore struct { @@ -123,9 +131,11 @@ func (s *gormStreamingAccountStore) Upsert(ctx context.Context, account Streamin } return tx.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "user_id"}, {Name: "provider"}}, + // reconnect_required_at 포함: 재연결이 곧 재연결 필요 상태의 해소다. DoUpdates: clause.AssignmentColumns([]string{ "channel_id", "channel_title", "refresh_token_ciphertext", "token_key_version", "refresh_token_expires_at", + "reconnect_required_at", "connected_at", "updated_at", }), }).Create(&account).Error @@ -178,6 +188,25 @@ func (s *gormStreamingAccountStore) UpdateStreamInfo(ctx context.Context, id uui }) } +func (s *gormStreamingAccountStore) MarkReconnectRequired(ctx context.Context, id uuid.UUID, at time.Time) error { + if s == nil || s.db == nil { + return errors.New("streaming account database is nil") + } + result := s.db.WithContext(ctx).Model(&StreamingAccount{}). + Where("id = ?", id). + Updates(map[string]any{ + "reconnect_required_at": at, + "updated_at": s.now(), + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return ErrStreamingAccountNotFound + } + return nil +} + func (s *gormStreamingAccountStore) UpdateRefreshToken(ctx context.Context, id uuid.UUID, ciphertext []byte, version *int16, expiresAt *time.Time) error { if s == nil || s.db == nil { return errors.New("streaming account database is nil") diff --git a/internal/auth/streaming_account_postgres_test.go b/internal/auth/streaming_account_postgres_test.go index bf12edb..c41c16a 100644 --- a/internal/auth/streaming_account_postgres_test.go +++ b/internal/auth/streaming_account_postgres_test.go @@ -94,6 +94,31 @@ func TestPostgresStreamingAccountUpsert(t *testing.T) { t.Fatalf("unknown id error = %v, want ErrStreamingAccountNotFound", err) } + // 재연결 필요 표식과 재연결(Upsert)에 의한 해소. + if err := store.MarkReconnectRequired(ctx, updated.ID, now); err != nil { + t.Fatal(err) + } + marked, err := store.Get(ctx, user.ID, StreamingProviderYouTube) + if err != nil { + t.Fatal(err) + } + if marked.ReconnectRequiredAt == nil { + t.Fatal("ReconnectRequiredAt not persisted") + } + if err := store.Upsert(ctx, StreamingAccount{UserID: user.ID, Provider: StreamingProviderYouTube, ChannelID: "UCsecond"}); err != nil { + t.Fatal(err) + } + cleared, err := store.Get(ctx, user.ID, StreamingProviderYouTube) + if err != nil { + t.Fatal(err) + } + if cleared.ReconnectRequiredAt != nil { + t.Fatalf("ReconnectRequiredAt = %v after reconnect, want nil", cleared.ReconnectRequiredAt) + } + if err := store.MarkReconnectRequired(ctx, uuid.New(), now); !errors.Is(err, ErrStreamingAccountNotFound) { + t.Fatalf("unknown id mark error = %v, want ErrStreamingAccountNotFound", err) + } + // 비활성 사용자의 연결은 거부돼야 한다. disabled := User{ID: uuid.New(), Status: UserStatusDisabled, CreatedAt: now, UpdatedAt: now} if err := db.Create(&disabled).Error; err != nil { diff --git a/internal/auth/youtube_oauth.go b/internal/auth/youtube_oauth.go index ce06bf3..2942254 100644 --- a/internal/auth/youtube_oauth.go +++ b/internal/auth/youtube_oauth.go @@ -35,6 +35,10 @@ var ( ErrYouTubeTokenExchange = errors.New("YouTube token exchange failed") ErrYouTubeAuthCodeRejected = errors.New("YouTube authorization code was rejected") ErrStreamingNotConnected = errors.New("streaming account is not connected") + // ErrStreamingReconnectRequired: 저장된 refresh token이 무효화됨(사용자의 + // 플랫폼 쪽 권한 취소 등). 재시도로 복구되지 않으며 재연결이 유일한 해법 + // 이라 일반 실패와 구분한다. + ErrStreamingReconnectRequired = errors.New("streaming account requires reconnection") ) // CodeSource는 인가 코드를 발급받은 클라이언트 유형이다. 교환 시 요구되는 @@ -387,6 +391,13 @@ func (p *YouTubeAccessTokenProvider) AccessToken(ctx context.Context, userID uui } response, err := p.oauth.RefreshAccessToken(ctx, refreshToken) if err != nil { + // 토큰 엔드포인트의 4xx는 refresh token 자체가 무효라는 뜻이다(만료· + // 권한 취소). 재연결 필요로 표식하고 전용 에러로 구분한다 — 표식 + // 실패는 무시한다(다음 시도에서 다시 표식된다). + if errors.Is(err, ErrYouTubeAuthCodeRejected) { + _ = p.store.MarkReconnectRequired(ctx, account.ID, p.now()) + return "", fmt.Errorf("%w: %v", ErrStreamingReconnectRequired, err) + } return "", err } // Google이 예외적으로 새 refresh token을 주면 행 락 하에 교체 저장한다. diff --git a/internal/auth/youtube_oauth_test.go b/internal/auth/youtube_oauth_test.go index a065ed2..30895f9 100644 --- a/internal/auth/youtube_oauth_test.go +++ b/internal/auth/youtube_oauth_test.go @@ -289,6 +289,19 @@ func (s *memoryStreamingAccountStore) UpdateRefreshToken(_ context.Context, id u return ErrStreamingAccountNotFound } +func (s *memoryStreamingAccountStore) MarkReconnectRequired(_ context.Context, id uuid.UUID, at time.Time) error { + s.mu.Lock() + defer s.mu.Unlock() + for key, account := range s.accounts { + if account.ID == id { + account.ReconnectRequiredAt = &at + s.accounts[key] = account + return nil + } + } + return ErrStreamingAccountNotFound +} + func (s *memoryStreamingAccountStore) UpdateStreamInfo(_ context.Context, id uuid.UUID, info StreamInfo) error { s.mu.Lock() defer s.mu.Unlock() @@ -462,6 +475,61 @@ func TestYouTubeAccessTokenProviderSerializesRefresh(t *testing.T) { } } +// TestYouTubeAccessTokenProviderMarksReconnectRequired: refresh token이 토큰 +// 엔드포인트에서 4xx로 거절되면(권한 취소·만료) 전용 에러로 구분되고 계정에 +// 재연결 필요 표식이 남아야 한다 — 조회 API가 API 호출 없이 판별하는 근거. +func TestYouTubeAccessTokenProviderMarksReconnectRequired(t *testing.T) { + cipher := testProviderTokenCipher(t) + ciphertext, version, err := cipher.Encrypt("rt-revoked") + if err != nil { + t.Fatal(err) + } + store := newMemoryStreamingAccountStore() + userID := uuid.New() + if err := store.Upsert(context.Background(), StreamingAccount{ + UserID: userID, + Provider: StreamingProviderYouTube, + ChannelID: "UCabc", + RefreshTokenCiphertext: ciphertext, + TokenKeyVersion: version, + }); err != nil { + t.Fatal(err) + } + oauth := &stubYouTubeAuthorizer{refreshErr: ErrYouTubeAuthCodeRejected} + provider, err := NewYouTubeAccessTokenProvider(oauth, store, cipher) + if err != nil { + t.Fatal(err) + } + if _, err := provider.AccessToken(context.Background(), userID); !errors.Is(err, ErrStreamingReconnectRequired) { + t.Fatalf("error = %v, want ErrStreamingReconnectRequired", err) + } + account, err := store.Get(context.Background(), userID, StreamingProviderYouTube) + if err != nil { + t.Fatal(err) + } + if account.ReconnectRequiredAt == nil { + t.Fatal("ReconnectRequiredAt must be marked after an invalid refresh token") + } + + // 재연결(Upsert)이 표식을 해소해야 한다. + if err := store.Upsert(context.Background(), StreamingAccount{ + UserID: userID, + Provider: StreamingProviderYouTube, + ChannelID: "UCabc", + RefreshTokenCiphertext: ciphertext, + TokenKeyVersion: version, + }); err != nil { + t.Fatal(err) + } + account, err = store.Get(context.Background(), userID, StreamingProviderYouTube) + if err != nil { + t.Fatal(err) + } + if account.ReconnectRequiredAt != nil { + t.Fatalf("ReconnectRequiredAt = %v after reconnect, want nil", account.ReconnectRequiredAt) + } +} + func TestYouTubeAccessTokenProviderNotConnected(t *testing.T) { provider, err := NewYouTubeAccessTokenProvider(&stubYouTubeAuthorizer{}, newMemoryStreamingAccountStore(), testProviderTokenCipher(t)) if err != nil { diff --git a/internal/database/migration/sql/000005_add_streaming_reconnect_required.down.sql b/internal/database/migration/sql/000005_add_streaming_reconnect_required.down.sql new file mode 100644 index 0000000..a10c662 --- /dev/null +++ b/internal/database/migration/sql/000005_add_streaming_reconnect_required.down.sql @@ -0,0 +1,6 @@ +BEGIN; + +ALTER TABLE streaming_accounts + DROP COLUMN IF EXISTS reconnect_required_at; + +COMMIT; diff --git a/internal/database/migration/sql/000005_add_streaming_reconnect_required.up.sql b/internal/database/migration/sql/000005_add_streaming_reconnect_required.up.sql new file mode 100644 index 0000000..b9f54cc --- /dev/null +++ b/internal/database/migration/sql/000005_add_streaming_reconnect_required.up.sql @@ -0,0 +1,6 @@ +BEGIN; + +ALTER TABLE streaming_accounts + ADD COLUMN reconnect_required_at TIMESTAMPTZ; + +COMMIT; diff --git a/internal/server/server.go b/internal/server/server.go index 0b22ee1..5a4518c 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -212,6 +212,10 @@ func (s *Server) handleStartStream(w http.ResponseWriter, r *http.Request, liveS switch { case errors.Is(err, auth.ErrStreamingNotConnected): writeError(w, apiError{Status: http.StatusConflict, Code: "streaming_not_connected", Message: "Connect a streaming account before starting a stream.", Details: map[string]any{"provider": providerName}}) + case errors.Is(err, auth.ErrStreamingReconnectRequired): + // 재시도로 복구되지 않는 상태 — "잠시 후 재시도"가 아니라 + // "재연결"을 안내해야 하므로 일반 준비 실패(502)와 구분한다. + writeError(w, apiError{Status: http.StatusConflict, Code: "streaming_reconnect_required", Message: "The streaming account needs to be reconnected.", Details: map[string]any{"provider": providerName}}) case errors.Is(err, streaming.ErrLiveStreamingBlocked): writeError(w, apiError{Status: http.StatusForbidden, Code: "live_streaming_blocked", Message: "The channel is not enabled for live streaming. Enabling can take up to 24 hours.", Details: map[string]any{"help_url": streaming.LiveStreamingHelpURL}}) default: diff --git a/internal/server/stream_test.go b/internal/server/stream_test.go index 44abe09..d8f15e3 100644 --- a/internal/server/stream_test.go +++ b/internal/server/stream_test.go @@ -122,6 +122,13 @@ func TestStartStreamMapsProviderErrors(t *testing.T) { wantStatus: http.StatusForbidden, wantCode: "live_streaming_blocked", }, + { + // 무효 refresh token은 재시도가 아니라 재연결 안내여야 한다(#88). + name: "reconnect required", + provider: &stubStreamingProvider{prepareErr: auth.ErrStreamingReconnectRequired}, + wantStatus: http.StatusConflict, + wantCode: "streaming_reconnect_required", + }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/internal/streaming/youtube_test.go b/internal/streaming/youtube_test.go index 55ef621..d8ba7ae 100644 --- a/internal/streaming/youtube_test.go +++ b/internal/streaming/youtube_test.go @@ -68,6 +68,18 @@ func (s *memoryStore) UpdateRefreshToken(_ context.Context, id uuid.UUID, cipher return nil } +func (s *memoryStore) MarkReconnectRequired(_ context.Context, id uuid.UUID, at time.Time) error { + s.mu.Lock() + defer s.mu.Unlock() + account, ok := s.accounts[id] + if !ok { + return auth.ErrStreamingAccountNotFound + } + account.ReconnectRequiredAt = &at + s.accounts[id] = account + return nil +} + func (s *memoryStore) UpdateStreamInfo(_ context.Context, id uuid.UUID, info auth.StreamInfo) error { s.mu.Lock() defer s.mu.Unlock() From 9010408d71c262ce7724ada0c710001f2547f16d Mon Sep 17 00:00:00 2001 From: itzjb Date: Mon, 10 Aug 2026 19:53:18 +0900 Subject: [PATCH 2/4] =?UTF-8?q?feat:=20=EC=86=A1=EC=B6=9C=20=EA=B3=84?= =?UTF-8?q?=EC=A0=95=20=EC=97=B0=EA=B2=B0=20=EB=AA=A9=EB=A1=9D=20=EC=A1=B0?= =?UTF-8?q?=ED=9A=8C=20API=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/server/main.go | 11 +- internal/auth/streaming_account.go | 16 ++ internal/auth/streaming_account_http.go | 57 +++++++ internal/auth/streaming_account_http_test.go | 166 +++++++++++++++++++ internal/auth/streaming_account_service.go | 89 ++++++++++ internal/auth/token_http.go | 36 ++-- internal/auth/youtube_oauth_test.go | 15 ++ internal/streaming/youtube_test.go | 18 +- 8 files changed, 390 insertions(+), 18 deletions(-) create mode 100644 internal/auth/streaming_account_http.go create mode 100644 internal/auth/streaming_account_http_test.go create mode 100644 internal/auth/streaming_account_service.go diff --git a/cmd/server/main.go b/cmd/server/main.go index 630504e..d196f16 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -320,6 +320,14 @@ func main() { // 배포에서도 refresh token 암호화가 성립해야 한다. var youtubeConnect *auth.YouTubeConnectService streamingProviders := map[auth.StreamingProvider]streaming.Provider{} + // 송출 계정 저장소·조회 서비스는 플랫폼 중립이라 YouTube 설정 여부와 + // 무관하게 조립한다 — 연결이 없으면 조회가 빈 배열을 돌려줄 뿐이다. + streamingAccountStore := auth.NewGormStreamingAccountStore(databaseConnection.DB) + streamingAccounts, err := auth.NewStreamingAccountService(streamingAccountStore, userStatusChecker) + if err != nil { + logger.Error("create streaming account service failed", "error", err) + os.Exit(2) + } if youtubeOAuthConfig.Enabled() { if providerTokenCipher == nil { providerTokenCipher, err = auth.NewProviderTokenCipherFromBase64(os.Getenv("AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64")) @@ -333,7 +341,6 @@ func main() { logger.Error("create YouTube OAuth client failed", "error", err) os.Exit(2) } - streamingAccountStore := auth.NewGormStreamingAccountStore(databaseConnection.DB) youtubeConnect, err = auth.NewYouTubeConnectService( youtubeOAuthClient, streamingAccountStore, @@ -422,7 +429,7 @@ func main() { httpServer := &http.Server{ Addr: cfg.HTTPAddr, - Handler: auth.MountAuthHTTPWithServices(application.Handler(), tokenService, googleLogin, appleLogin, emailLogin, withdrawal, logger, originConfig, youtubeConnect), + Handler: auth.MountAuthHTTPWithStreaming(application.Handler(), tokenService, googleLogin, appleLogin, emailLogin, withdrawal, youtubeConnect, streamingAccounts, logger, originConfig), ReadHeaderTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second, } diff --git a/internal/auth/streaming_account.go b/internal/auth/streaming_account.go index db56043..8ad57f2 100644 --- a/internal/auth/streaming_account.go +++ b/internal/auth/streaming_account.go @@ -95,6 +95,8 @@ type StreamingAccountStore interface { // 토큰을 교체하고 기존 행(ID)을 유지한다. Upsert(ctx context.Context, account StreamingAccount) error Get(ctx context.Context, userID uuid.UUID, provider StreamingProvider) (StreamingAccount, error) + // ListByUser는 사용자의 모든 플랫폼 연결을 provider 순으로 돌려준다. + ListByUser(ctx context.Context, userID uuid.UUID) ([]StreamingAccount, error) // UpdateRefreshToken은 토큰 갱신 응답이 새 refresh token을 담아온 경우 // 행 락 하에 교체한다 — 한 사용자의 다중 세션이 동시에 갱신할 때 나중에 // 실패한 쓰기가 최신 토큰을 덮지 않도록 잠근다. @@ -188,6 +190,20 @@ func (s *gormStreamingAccountStore) UpdateStreamInfo(ctx context.Context, id uui }) } +func (s *gormStreamingAccountStore) ListByUser(ctx context.Context, userID uuid.UUID) ([]StreamingAccount, error) { + if s == nil || s.db == nil { + return nil, errors.New("streaming account database is nil") + } + var accounts []StreamingAccount + if err := s.db.WithContext(ctx). + Where("user_id = ?", userID). + Order("provider ASC"). + Find(&accounts).Error; err != nil { + return nil, err + } + return accounts, nil +} + func (s *gormStreamingAccountStore) MarkReconnectRequired(ctx context.Context, id uuid.UUID, at time.Time) error { if s == nil || s.db == nil { return errors.New("streaming account database is nil") diff --git a/internal/auth/streaming_account_http.go b/internal/auth/streaming_account_http.go new file mode 100644 index 0000000..e4d3f91 --- /dev/null +++ b/internal/auth/streaming_account_http.go @@ -0,0 +1,57 @@ +package auth + +import ( + "net/http" + + "github.com/google/uuid" +) + +// authenticatedUserID는 Bearer 액세스 토큰에서 사용자 UUID를 복원한다. +// handleWithdrawal의 인라인 인증 3단계와 동일한 패턴이다 — 사용자 상태(active) +// 확인은 서비스 계층이 담당한다. +func (h *tokenHTTPHandler) authenticatedUserID(w http.ResponseWriter, r *http.Request) (uuid.UUID, bool) { + raw, ok := accessBearerToken(r) + if !ok { + h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.") + return uuid.Nil, false + } + claims, err := h.service.ValidateAccessToken(raw) + if err != nil { + h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.") + return uuid.Nil, false + } + userID, err := uuid.Parse(claims.Subject) + if err != nil { + h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.") + return uuid.Nil, false + } + return userID, true +} + +// handleListStreamingAccounts는 사용자의 플랫폼 연결 목록을 돌려준다. +// 플랫폼 중립 엔드포인트다 — 새 플랫폼은 배열 항목으로만 나타난다(#88). +func (h *tokenHTTPHandler) handleListStreamingAccounts(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + userID, ok := h.authenticatedUserID(w, r) + if !ok { + return + } + summaries, err := h.streamingAccounts.List(r.Context(), userID) + if err != nil { + if isUnauthorizedStreamingError(err) { + h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.") + return + } + h.logger.Error("list streaming accounts failed", "request_id", tokenRequestID(r), "error", err) + h.writeError(w, r, http.StatusInternalServerError, "internal_error", "An unexpected server error occurred.") + return + } + h.writeJSON(w, http.StatusOK, summaries) +} + +func isUnauthorizedStreamingError(err error) bool { + return err == ErrUserInactive +} diff --git a/internal/auth/streaming_account_http_test.go b/internal/auth/streaming_account_http_test.go new file mode 100644 index 0000000..393e1ed --- /dev/null +++ b/internal/auth/streaming_account_http_test.go @@ -0,0 +1,166 @@ +package auth + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/google/uuid" +) + +func testStreamingAccountsHandler(t *testing.T, store StreamingAccountStore, status UserStatus) (*TokenService, http.Handler) { + t.Helper() + config, err := NewTokenHTTPConfig(false, nil) + if err != nil { + t.Fatal(err) + } + tokens := testTokenService(newMemoryRefreshStore()) + service, err := NewStreamingAccountService(store, testUserStatusChecker{status: status}) + if err != nil { + t.Fatal(err) + } + handler := MountAuthHTTPWithStreaming(http.NotFoundHandler(), tokens, nil, nil, nil, nil, nil, service, slog.New(slog.NewTextHandler(io.Discard, nil)), config) + return tokens, handler +} + +func listStreamingAccounts(t *testing.T, handler http.Handler, accessToken string) *httptest.ResponseRecorder { + t.Helper() + request := httptest.NewRequest(http.MethodGet, "/auth/streaming/accounts", nil) + if accessToken != "" { + request.Header.Set("Authorization", "Bearer "+accessToken) + } + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + return response +} + +func TestListStreamingAccountsRequiresBearer(t *testing.T) { + _, handler := testStreamingAccountsHandler(t, newMemoryStreamingAccountStore(), UserStatusActive) + if response := listStreamingAccounts(t, handler, ""); response.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401", response.Code) + } +} + +// TestListStreamingAccountsEmptyIsArray: 연결이 없으면 404가 아니라 빈 배열 +// (JSON `[]`, null 아님)이어야 한다 — 이슈 #88 계약. +func TestListStreamingAccountsEmptyIsArray(t *testing.T) { + tokens, handler := testStreamingAccountsHandler(t, newMemoryStreamingAccountStore(), UserStatusActive) + pair, err := tokens.IssuePair(context.Background(), uuid.New(), ClientInfo{}) + if err != nil { + t.Fatal(err) + } + response := listStreamingAccounts(t, handler, pair.AccessToken) + if response.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", response.Code, response.Body.String()) + } + if body := strings.TrimSpace(response.Body.String()); body != "[]" { + t.Fatalf("empty list body = %q, want []", body) + } +} + +func TestListStreamingAccountsReturnsConnections(t *testing.T) { + store := newMemoryStreamingAccountStore() + userID := uuid.New() + title := "Team Framework" + if err := store.Upsert(context.Background(), StreamingAccount{ + UserID: userID, + Provider: StreamingProviderYouTube, + ChannelID: "UCabc", + ChannelTitle: &title, + }); err != nil { + t.Fatal(err) + } + // 다른 사용자의 연결은 보이면 안 된다. + if err := store.Upsert(context.Background(), StreamingAccount{ + UserID: uuid.New(), + Provider: StreamingProviderYouTube, + ChannelID: "UCother", + }); err != nil { + t.Fatal(err) + } + + tokens, handler := testStreamingAccountsHandler(t, store, UserStatusActive) + pair, err := tokens.IssuePair(context.Background(), userID, ClientInfo{}) + if err != nil { + t.Fatal(err) + } + response := listStreamingAccounts(t, handler, pair.AccessToken) + if response.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", response.Code, response.Body.String()) + } + var payload []struct { + Provider string `json:"provider"` + ChannelID string `json:"channel_id"` + ChannelTitle string `json:"channel_title"` + ConnectedAt string `json:"connected_at"` + ReconnectRequired bool `json:"reconnect_required"` + } + if err := json.NewDecoder(response.Body).Decode(&payload); err != nil { + t.Fatal(err) + } + if len(payload) != 1 { + t.Fatalf("items = %d, want 1 (own connection only)", len(payload)) + } + item := payload[0] + if item.Provider != "youtube" || item.ChannelID != "UCabc" || item.ChannelTitle != "Team Framework" { + t.Fatalf("item = %+v", item) + } + if item.ReconnectRequired { + t.Fatal("healthy connection must not require reconnect") + } + if item.ConnectedAt == "" { + t.Fatal("connected_at missing") + } +} + +// TestListStreamingAccountsReconnectRequired: 표식·만료 어느 쪽으로든 재연결 +// 필요가 조회 응답에 드러나야 한다 — 플랫폼 API 호출 없이. +func TestListStreamingAccountsReconnectRequired(t *testing.T) { + now := time.Now().UTC() + past := now.Add(-time.Hour) + + cases := []struct { + name string + mutate func(*StreamingAccount) + require bool + }{ + {"marked invalid", func(a *StreamingAccount) { a.ReconnectRequiredAt = &past }, true}, + {"testing token expired", func(a *StreamingAccount) { a.RefreshTokenExpiresAt = &past }, true}, + {"testing token still valid", func(a *StreamingAccount) { + future := now.Add(24 * time.Hour) + a.RefreshTokenExpiresAt = &future + }, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + store := newMemoryStreamingAccountStore() + userID := uuid.New() + account := StreamingAccount{UserID: userID, Provider: StreamingProviderYouTube, ChannelID: "UCabc"} + tc.mutate(&account) + if err := store.Upsert(context.Background(), account); err != nil { + t.Fatal(err) + } + tokens, handler := testStreamingAccountsHandler(t, store, UserStatusActive) + pair, err := tokens.IssuePair(context.Background(), userID, ClientInfo{}) + if err != nil { + t.Fatal(err) + } + response := listStreamingAccounts(t, handler, pair.AccessToken) + var payload []struct { + ReconnectRequired bool `json:"reconnect_required"` + } + if err := json.NewDecoder(response.Body).Decode(&payload); err != nil { + t.Fatal(err) + } + if len(payload) != 1 || payload[0].ReconnectRequired != tc.require { + t.Fatalf("payload = %+v, want reconnect_required=%v", payload, tc.require) + } + }) + } +} diff --git a/internal/auth/streaming_account_service.go b/internal/auth/streaming_account_service.go new file mode 100644 index 0000000..d929b1f --- /dev/null +++ b/internal/auth/streaming_account_service.go @@ -0,0 +1,89 @@ +package auth + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" +) + +// StreamingAccountSummary는 연결 목록 조회 API의 응답 항목이다. 플랫폼 중립 +// 형태라 치지직이 붙어도 배열 항목만 늘어난다(#88). +type StreamingAccountSummary struct { + Provider StreamingProvider `json:"provider"` + ChannelID string `json:"channel_id"` + ChannelTitle string `json:"channel_title"` + ConnectedAt time.Time `json:"connected_at"` + // ReconnectRequired는 저장된 표식(reconnect_required_at)과 refresh token + // 만료 시각만으로 판별한다 — 조회마다 플랫폼 API를 부르지 않는다. + ReconnectRequired bool `json:"reconnect_required"` +} + +// StreamingAccountService는 송출 계정 연결의 조회·해제를 담당한다. +type StreamingAccountService struct { + store StreamingAccountStore + users UserStatusChecker + now func() time.Time +} + +func NewStreamingAccountService(store StreamingAccountStore, users UserStatusChecker) (*StreamingAccountService, error) { + if store == nil || users == nil { + return nil, errors.New("streaming account service dependencies must not be nil") + } + return &StreamingAccountService{ + store: store, + users: users, + now: func() time.Time { return time.Now().UTC() }, + }, nil +} + +// List는 사용자의 플랫폼 연결 목록을 돌려준다. 연결이 없으면 빈 슬라이스다 +// (404가 아니라 빈 배열 — 이슈 #88 계약). +func (s *StreamingAccountService) List(ctx context.Context, userID uuid.UUID) ([]StreamingAccountSummary, error) { + if err := s.ensureActive(ctx, userID); err != nil { + return nil, err + } + accounts, err := s.store.ListByUser(ctx, userID) + if err != nil { + return nil, err + } + // make로 시작해 JSON이 null이 아니라 []로 직렬화되게 한다. + summaries := make([]StreamingAccountSummary, 0, len(accounts)) + for _, account := range accounts { + summary := StreamingAccountSummary{ + Provider: account.Provider, + ChannelID: account.ChannelID, + ConnectedAt: account.ConnectedAt, + ReconnectRequired: s.reconnectRequired(account), + } + if account.ChannelTitle != nil { + summary.ChannelTitle = *account.ChannelTitle + } + summaries = append(summaries, summary) + } + return summaries, nil +} + +func (s *StreamingAccountService) reconnectRequired(account StreamingAccount) bool { + if account.ReconnectRequiredAt != nil { + return true + } + // Testing 게시 상태 시절 발급된 토큰의 만료(실측 7일). 만료가 지났으면 + // 갱신 시도가 실패할 것이 확정적이므로 시도 전에 재연결로 안내한다. + if account.RefreshTokenExpiresAt != nil && !s.now().Before(*account.RefreshTokenExpiresAt) { + return true + } + return false +} + +func (s *StreamingAccountService) ensureActive(ctx context.Context, userID uuid.UUID) error { + status, err := s.users.UserStatus(ctx, userID) + if err != nil { + return err + } + if status != UserStatusActive { + return ErrUserInactive + } + return nil +} diff --git a/internal/auth/token_http.go b/internal/auth/token_http.go index 686517d..fddf66f 100644 --- a/internal/auth/token_http.go +++ b/internal/auth/token_http.go @@ -17,14 +17,15 @@ import ( const maxTokenRequestBody = 8 << 10 type tokenHTTPHandler struct { - service *TokenService - google *GoogleLoginService - apple *AppleLoginService - email *EmailAuthService - withdrawal *AccountWithdrawalService - youtube *YouTubeConnectService - logger *slog.Logger - config TokenHTTPConfig + service *TokenService + google *GoogleLoginService + apple *AppleLoginService + email *EmailAuthService + withdrawal *AccountWithdrawalService + youtube *YouTubeConnectService + streamingAccounts *StreamingAccountService + logger *slog.Logger + config TokenHTTPConfig } func MountTokenHTTP(next http.Handler, service *TokenService, logger *slog.Logger, config TokenHTTPConfig) http.Handler { @@ -36,14 +37,14 @@ func MountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLogin if len(appleServices) > 0 { apple = appleServices[0] } - return mountAuthHTTP(next, service, google, apple, nil, nil, nil, logger, config) + return mountAuthHTTP(next, service, google, apple, nil, nil, nil, nil, logger, config) } // MountAuthHTTPWithWithdrawal adds account deletion to the authentication // routes. It is kept separate to preserve the existing constructor used by // smaller deployments and focused handler tests. func MountAuthHTTPWithWithdrawal(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, withdrawal *AccountWithdrawalService, logger *slog.Logger, config TokenHTTPConfig) http.Handler { - return mountAuthHTTP(next, service, google, apple, nil, withdrawal, nil, logger, config) + return mountAuthHTTP(next, service, google, apple, nil, withdrawal, nil, nil, logger, config) } // MountAuthHTTPWithServices mounts all configured authentication services. @@ -54,14 +55,20 @@ func MountAuthHTTPWithServices(next http.Handler, service *TokenService, google if len(youtubeServices) > 0 { youtube = youtubeServices[0] } - return mountAuthHTTP(next, service, google, apple, email, withdrawal, youtube, logger, config) + return mountAuthHTTP(next, service, google, apple, email, withdrawal, youtube, nil, logger, config) } -func mountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, email *EmailAuthService, withdrawal *AccountWithdrawalService, youtube *YouTubeConnectService, logger *slog.Logger, config TokenHTTPConfig) http.Handler { +// MountAuthHTTPWithStreaming은 송출 계정 조회·해제(#88)까지 포함해 전체 +// 인증 라우트를 마운트한다 — 프로덕션 조립(main)이 쓰는 완전형이다. +func MountAuthHTTPWithStreaming(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, email *EmailAuthService, withdrawal *AccountWithdrawalService, youtube *YouTubeConnectService, streamingAccounts *StreamingAccountService, logger *slog.Logger, config TokenHTTPConfig) http.Handler { + return mountAuthHTTP(next, service, google, apple, email, withdrawal, youtube, streamingAccounts, logger, config) +} + +func mountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, email *EmailAuthService, withdrawal *AccountWithdrawalService, youtube *YouTubeConnectService, streamingAccounts *StreamingAccountService, logger *slog.Logger, config TokenHTTPConfig) http.Handler { if logger == nil { logger = slog.Default() } - h := &tokenHTTPHandler{service: service, google: google, apple: apple, email: email, withdrawal: withdrawal, youtube: youtube, logger: logger, config: config} + h := &tokenHTTPHandler{service: service, google: google, apple: apple, email: email, withdrawal: withdrawal, youtube: youtube, streamingAccounts: streamingAccounts, logger: logger, config: config} mux := http.NewServeMux() mux.Handle("/auth/refresh", h.middleware(http.HandlerFunc(h.handleRefresh))) mux.Handle("/auth/logout", h.middleware(http.HandlerFunc(h.handleLogout))) @@ -85,6 +92,9 @@ func mountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLogin mux.Handle("POST /auth/youtube/connect", h.middleware(http.HandlerFunc(h.handleYouTubeConnect))) mux.Handle("GET /auth/youtube/config", h.middleware(http.HandlerFunc(h.handleYouTubeConfig))) } + if h.streamingAccounts != nil { + mux.Handle("GET /auth/streaming/accounts", h.middleware(http.HandlerFunc(h.handleListStreamingAccounts))) + } mux.Handle("/", next) return mux } diff --git a/internal/auth/youtube_oauth_test.go b/internal/auth/youtube_oauth_test.go index 30895f9..8cab26f 100644 --- a/internal/auth/youtube_oauth_test.go +++ b/internal/auth/youtube_oauth_test.go @@ -260,10 +260,25 @@ func (s *memoryStreamingAccountStore) Upsert(_ context.Context, account Streamin } else if account.ID == uuid.Nil { account.ID = uuid.New() } + if account.ConnectedAt.IsZero() { + account.ConnectedAt = time.Now().UTC() + } s.accounts[key] = account return nil } +func (s *memoryStreamingAccountStore) ListByUser(_ context.Context, userID uuid.UUID) ([]StreamingAccount, error) { + s.mu.Lock() + defer s.mu.Unlock() + var accounts []StreamingAccount + for _, account := range s.accounts { + if account.UserID == userID { + accounts = append(accounts, account) + } + } + return accounts, nil +} + func (s *memoryStreamingAccountStore) Get(_ context.Context, userID uuid.UUID, provider StreamingProvider) (StreamingAccount, error) { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/streaming/youtube_test.go b/internal/streaming/youtube_test.go index d8ba7ae..51ec4e1 100644 --- a/internal/streaming/youtube_test.go +++ b/internal/streaming/youtube_test.go @@ -43,6 +43,18 @@ func (s *memoryStore) Upsert(_ context.Context, account auth.StreamingAccount) e return nil } +func (s *memoryStore) ListByUser(_ context.Context, userID uuid.UUID) ([]auth.StreamingAccount, error) { + s.mu.Lock() + defer s.mu.Unlock() + var accounts []auth.StreamingAccount + for _, account := range s.accounts { + if account.UserID == userID { + accounts = append(accounts, account) + } + } + return accounts, nil +} + func (s *memoryStore) Get(_ context.Context, userID uuid.UUID, provider auth.StreamingProvider) (auth.StreamingAccount, error) { s.mu.Lock() defer s.mu.Unlock() @@ -173,9 +185,9 @@ func testProviderWith(t *testing.T, stub *youtubeAPIStub, store auth.StreamingAc func connectedAccount(t *testing.T, store *memoryStore, userID uuid.UUID) auth.StreamingAccount { t.Helper() account := auth.StreamingAccount{ - ID: uuid.New(), - UserID: userID, - Provider: auth.StreamingProviderYouTube, + ID: uuid.New(), + UserID: userID, + Provider: auth.StreamingProviderYouTube, ChannelID: "UCabc", } if err := store.Upsert(context.Background(), account); err != nil { From 3134fbb2b92ed7e16527b0a0e3f994640083dd1b Mon Sep 17 00:00:00 2001 From: itzjb Date: Mon, 10 Aug 2026 19:56:41 +0900 Subject: [PATCH 3/4] =?UTF-8?q?feat:=20=EC=86=A1=EC=B6=9C=20=EA=B3=84?= =?UTF-8?q?=EC=A0=95=20=EC=97=B0=EA=B2=B0=20=ED=95=B4=EC=A0=9C=20API=20?= =?UTF-8?q?=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/server/main.go | 17 +- internal/auth/streaming_account.go | 16 ++ internal/auth/streaming_account_http.go | 30 +++ internal/auth/streaming_account_http_test.go | 44 +++- .../auth/streaming_account_postgres_test.go | 21 ++ internal/auth/streaming_account_service.go | 69 +++++- .../auth/streaming_account_service_test.go | 198 ++++++++++++++++++ internal/auth/token_http.go | 1 + internal/auth/youtube_oauth.go | 30 ++- internal/auth/youtube_oauth_test.go | 12 ++ internal/streaming/youtube.go | 30 +++ internal/streaming/youtube_test.go | 49 +++++ 12 files changed, 502 insertions(+), 15 deletions(-) create mode 100644 internal/auth/streaming_account_service_test.go diff --git a/cmd/server/main.go b/cmd/server/main.go index d196f16..0146258 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -323,11 +323,7 @@ func main() { // 송출 계정 저장소·조회 서비스는 플랫폼 중립이라 YouTube 설정 여부와 // 무관하게 조립한다 — 연결이 없으면 조회가 빈 배열을 돌려줄 뿐이다. streamingAccountStore := auth.NewGormStreamingAccountStore(databaseConnection.DB) - streamingAccounts, err := auth.NewStreamingAccountService(streamingAccountStore, userStatusChecker) - if err != nil { - logger.Error("create streaming account service failed", "error", err) - os.Exit(2) - } + streamingDisconnectHooks := map[auth.StreamingProvider]auth.StreamingDisconnectHooks{} if youtubeOAuthConfig.Enabled() { if providerTokenCipher == nil { providerTokenCipher, err = auth.NewProviderTokenCipherFromBase64(os.Getenv("AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64")) @@ -362,6 +358,17 @@ func main() { os.Exit(2) } streamingProviders[auth.StreamingProviderYouTube] = youtubeProvider + // 해제 시 정리 훅: ①재사용 스트림 삭제(Live API) ②Google 권한 취소. + streamingDisconnectHooks[auth.StreamingProviderYouTube] = auth.StreamingDisconnectHooks{ + CleanupResources: youtubeProvider.CleanupStreamingResources, + RevokeToken: youtubeOAuthClient.RevokeToken, + } + } + // 조회·해제 서비스는 플랫폼 중립이라 훅 구성 뒤 한 번만 조립한다. + streamingAccounts, err := auth.NewStreamingAccountService(streamingAccountStore, userStatusChecker, providerTokenCipher, streamingDisconnectHooks, logger) + if err != nil { + logger.Error("create streaming account service failed", "error", err) + os.Exit(2) } // INNOLIVE_REQUIRE_SESSION_AUTH=false is the explicit local-development // escape hatch (loud warning above). Extend it to user auth as well so diff --git a/internal/auth/streaming_account.go b/internal/auth/streaming_account.go index 8ad57f2..d5a1b46 100644 --- a/internal/auth/streaming_account.go +++ b/internal/auth/streaming_account.go @@ -105,6 +105,8 @@ type StreamingAccountStore interface { UpdateStreamInfo(ctx context.Context, id uuid.UUID, info StreamInfo) error // MarkReconnectRequired는 토큰 갱신이 무효 토큰으로 거절됐음을 기록한다. MarkReconnectRequired(ctx context.Context, id uuid.UUID, at time.Time) error + // Delete는 연결 행을 삭제한다. 없으면 ErrStreamingAccountNotFound. + Delete(ctx context.Context, id uuid.UUID) error } type gormStreamingAccountStore struct { @@ -190,6 +192,20 @@ func (s *gormStreamingAccountStore) UpdateStreamInfo(ctx context.Context, id uui }) } +func (s *gormStreamingAccountStore) Delete(ctx context.Context, id uuid.UUID) error { + if s == nil || s.db == nil { + return errors.New("streaming account database is nil") + } + result := s.db.WithContext(ctx).Where("id = ?", id).Delete(&StreamingAccount{}) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return ErrStreamingAccountNotFound + } + return nil +} + func (s *gormStreamingAccountStore) ListByUser(ctx context.Context, userID uuid.UUID) ([]StreamingAccount, error) { if s == nil || s.db == nil { return nil, errors.New("streaming account database is nil") diff --git a/internal/auth/streaming_account_http.go b/internal/auth/streaming_account_http.go index e4d3f91..8218f72 100644 --- a/internal/auth/streaming_account_http.go +++ b/internal/auth/streaming_account_http.go @@ -52,6 +52,36 @@ func (h *tokenHTTPHandler) handleListStreamingAccounts(w http.ResponseWriter, r h.writeJSON(w, http.StatusOK, summaries) } +// handleDisconnectStreamingAccount는 플랫폼 연결을 해제한다(#88). +// 정리 순서(리소스 삭제→권한 취소→행 삭제)와 실패 허용은 서비스가 보장한다. +func (h *tokenHTTPHandler) handleDisconnectStreamingAccount(w http.ResponseWriter, r *http.Request) { + userID, ok := h.authenticatedUserID(w, r) + if !ok { + return + } + provider := StreamingProvider(r.PathValue("provider")) + err := h.streamingAccounts.Disconnect(r.Context(), userID, provider) + if err != nil { + switch { + case isUnauthorizedStreamingError(err): + h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.") + case errorsIsStreamingNotFound(err): + // 미지 provider 문자열도 "연결 없음"으로 수렴한다 — 저장된 적이 + // 없는 이름이므로 계약상 같은 404다. + h.writeError(w, r, http.StatusNotFound, "not_found", "No connected streaming account for this provider.") + default: + h.logger.Error("disconnect streaming account failed", "request_id", tokenRequestID(r), "provider", provider, "error", err) + h.writeError(w, r, http.StatusInternalServerError, "internal_error", "An unexpected server error occurred.") + } + return + } + w.WriteHeader(http.StatusNoContent) +} + func isUnauthorizedStreamingError(err error) bool { return err == ErrUserInactive } + +func errorsIsStreamingNotFound(err error) bool { + return err == ErrStreamingAccountNotFound +} diff --git a/internal/auth/streaming_account_http_test.go b/internal/auth/streaming_account_http_test.go index 393e1ed..bd293fc 100644 --- a/internal/auth/streaming_account_http_test.go +++ b/internal/auth/streaming_account_http_test.go @@ -21,7 +21,7 @@ func testStreamingAccountsHandler(t *testing.T, store StreamingAccountStore, sta t.Fatal(err) } tokens := testTokenService(newMemoryRefreshStore()) - service, err := NewStreamingAccountService(store, testUserStatusChecker{status: status}) + service, err := NewStreamingAccountService(store, testUserStatusChecker{status: status}, nil, nil, nil) if err != nil { t.Fatal(err) } @@ -119,6 +119,48 @@ func TestListStreamingAccountsReturnsConnections(t *testing.T) { } } +func TestDisconnectStreamingAccountHTTP(t *testing.T) { + store := newMemoryStreamingAccountStore() + userID := uuid.New() + if err := store.Upsert(context.Background(), StreamingAccount{ + UserID: userID, + Provider: StreamingProviderYouTube, + ChannelID: "UCabc", + }); err != nil { + t.Fatal(err) + } + tokens, handler := testStreamingAccountsHandler(t, store, UserStatusActive) + pair, err := tokens.IssuePair(context.Background(), userID, ClientInfo{}) + if err != nil { + t.Fatal(err) + } + + deleteRequest := func(provider, token string) *httptest.ResponseRecorder { + request := httptest.NewRequest(http.MethodDelete, "/auth/streaming/accounts/"+provider, nil) + if token != "" { + request.Header.Set("Authorization", "Bearer "+token) + } + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + return response + } + + if response := deleteRequest("youtube", ""); response.Code != http.StatusUnauthorized { + t.Fatalf("no bearer status = %d, want 401", response.Code) + } + if response := deleteRequest("youtube", pair.AccessToken); response.Code != http.StatusNoContent { + t.Fatalf("disconnect status = %d, want 204 (body %s)", response.Code, response.Body.String()) + } + // 이미 해제됨 — 두 번째는 404. + if response := deleteRequest("youtube", pair.AccessToken); response.Code != http.StatusNotFound { + t.Fatalf("second disconnect status = %d, want 404", response.Code) + } + // 미지 provider 문자열도 "연결 없음"과 같은 404로 수렴한다. + if response := deleteRequest("soop", pair.AccessToken); response.Code != http.StatusNotFound { + t.Fatalf("unknown provider status = %d, want 404", response.Code) + } +} + // TestListStreamingAccountsReconnectRequired: 표식·만료 어느 쪽으로든 재연결 // 필요가 조회 응답에 드러나야 한다 — 플랫폼 API 호출 없이. func TestListStreamingAccountsReconnectRequired(t *testing.T) { diff --git a/internal/auth/streaming_account_postgres_test.go b/internal/auth/streaming_account_postgres_test.go index c41c16a..f0d5c4f 100644 --- a/internal/auth/streaming_account_postgres_test.go +++ b/internal/auth/streaming_account_postgres_test.go @@ -119,6 +119,27 @@ func TestPostgresStreamingAccountUpsert(t *testing.T) { t.Fatalf("unknown id mark error = %v, want ErrStreamingAccountNotFound", err) } + // 목록 조회와 삭제. + listed, err := store.ListByUser(ctx, user.ID) + if err != nil { + t.Fatal(err) + } + if len(listed) != 1 || listed[0].ID != updated.ID { + t.Fatalf("ListByUser = %d items, want the user's single connection", len(listed)) + } + if err := store.Delete(ctx, updated.ID); err != nil { + t.Fatal(err) + } + if _, err := store.Get(ctx, user.ID, StreamingProviderYouTube); !errors.Is(err, ErrStreamingAccountNotFound) { + t.Fatal("row must be gone after Delete") + } + if err := store.Delete(ctx, updated.ID); !errors.Is(err, ErrStreamingAccountNotFound) { + t.Fatalf("double delete error = %v, want ErrStreamingAccountNotFound", err) + } + if err := store.Upsert(ctx, StreamingAccount{UserID: user.ID, Provider: StreamingProviderYouTube, ChannelID: "UCsecond"}); err != nil { + t.Fatal(err) + } + // 비활성 사용자의 연결은 거부돼야 한다. disabled := User{ID: uuid.New(), Status: UserStatusDisabled, CreatedAt: now, UpdatedAt: now} if err := db.Create(&disabled).Error; err != nil { diff --git a/internal/auth/streaming_account_service.go b/internal/auth/streaming_account_service.go index d929b1f..77e6a5f 100644 --- a/internal/auth/streaming_account_service.go +++ b/internal/auth/streaming_account_service.go @@ -3,6 +3,7 @@ package auth import ( "context" "errors" + "log/slog" "time" "github.com/google/uuid" @@ -20,24 +21,78 @@ type StreamingAccountSummary struct { ReconnectRequired bool `json:"reconnect_required"` } +// StreamingDisconnectHooks는 연결 해제 시 수행할 플랫폼별 정리 동작이다. +// 훅이 없는 플랫폼(정리할 것이 없는 경우)은 해당 단계를 건너뛴다. +type StreamingDisconnectHooks struct { + // CleanupResources는 플랫폼에 만들어 둔 리소스(재사용 스트림 등)를 + // 삭제한다 — DB 행만 지우면 사용자 채널에 고아 리소스가 누적된다. + CleanupResources func(ctx context.Context, account StreamingAccount) error + // RevokeToken은 플랫폼 쪽 권한 부여를 취소한다. 인자는 refresh token + // 평문이다. + RevokeToken func(ctx context.Context, refreshToken string) error +} + // StreamingAccountService는 송출 계정 연결의 조회·해제를 담당한다. type StreamingAccountService struct { - store StreamingAccountStore - users UserStatusChecker - now func() time.Time + store StreamingAccountStore + users UserStatusChecker + cipher *ProviderTokenCipher + hooks map[StreamingProvider]StreamingDisconnectHooks + logger *slog.Logger + now func() time.Time } -func NewStreamingAccountService(store StreamingAccountStore, users UserStatusChecker) (*StreamingAccountService, error) { +// NewStreamingAccountService를 만든다. cipher와 hooks는 해제 시 플랫폼 정리 +// (토큰 복호화·revoke)에만 쓰이므로 nil이어도 조회·행 삭제는 동작한다. +func NewStreamingAccountService(store StreamingAccountStore, users UserStatusChecker, cipher *ProviderTokenCipher, hooks map[StreamingProvider]StreamingDisconnectHooks, logger *slog.Logger) (*StreamingAccountService, error) { if store == nil || users == nil { return nil, errors.New("streaming account service dependencies must not be nil") } + if logger == nil { + logger = slog.Default() + } return &StreamingAccountService{ - store: store, - users: users, - now: func() time.Time { return time.Now().UTC() }, + store: store, + users: users, + cipher: cipher, + hooks: hooks, + logger: logger, + now: func() time.Time { return time.Now().UTC() }, }, nil } +// Disconnect는 연결을 해제한다. 세 단계를 순서대로 수행한다(#88): +// ①플랫폼 리소스 삭제 → ②플랫폼 권한 취소 → ③DB 행 삭제. 토큰을 먼저 +// 폐기하면 ①을 못 하므로 순서를 바꾸면 안 되고, ①·②가 실패해도 ③은 +// 수행한다 — 이미 토큰이 무효화된 연결을 해제하는 것이 정상 시나리오다. +func (s *StreamingAccountService) Disconnect(ctx context.Context, userID uuid.UUID, provider StreamingProvider) error { + if err := s.ensureActive(ctx, userID); err != nil { + return err + } + account, err := s.store.Get(ctx, userID, provider) + if err != nil { + return err + } + hooks := s.hooks[provider] + if hooks.CleanupResources != nil { + if err := hooks.CleanupResources(ctx, account); err != nil { + s.logger.Warn("streaming resource cleanup failed; continuing disconnect", + "provider", provider, "user_id", userID, "error", err) + } + } + if hooks.RevokeToken != nil && s.cipher != nil && len(account.RefreshTokenCiphertext) > 0 { + refreshToken, err := s.cipher.Decrypt(account.RefreshTokenCiphertext, account.TokenKeyVersion) + if err != nil { + s.logger.Warn("streaming refresh token decrypt failed; skipping revoke", + "provider", provider, "user_id", userID, "error", err) + } else if err := hooks.RevokeToken(ctx, refreshToken); err != nil { + s.logger.Warn("streaming token revoke failed; continuing disconnect", + "provider", provider, "user_id", userID, "error", err) + } + } + return s.store.Delete(ctx, account.ID) +} + // List는 사용자의 플랫폼 연결 목록을 돌려준다. 연결이 없으면 빈 슬라이스다 // (404가 아니라 빈 배열 — 이슈 #88 계약). func (s *StreamingAccountService) List(ctx context.Context, userID uuid.UUID) ([]StreamingAccountSummary, error) { diff --git a/internal/auth/streaming_account_service_test.go b/internal/auth/streaming_account_service_test.go new file mode 100644 index 0000000..11ebbac --- /dev/null +++ b/internal/auth/streaming_account_service_test.go @@ -0,0 +1,198 @@ +package auth + +import ( + "context" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/google/uuid" +) + +// orderRecordingStore는 Disconnect의 3단계 순서를 기록하기 위한 store 래퍼다. +type orderRecordingStore struct { + StreamingAccountStore + mu *sync.Mutex + order *[]string +} + +func (s orderRecordingStore) Delete(ctx context.Context, id uuid.UUID) error { + s.mu.Lock() + *s.order = append(*s.order, "delete") + s.mu.Unlock() + return s.StreamingAccountStore.Delete(ctx, id) +} + +func disconnectFixture(t *testing.T, cleanupErr, revokeErr error) (*StreamingAccountService, *memoryStreamingAccountStore, uuid.UUID, *[]string) { + t.Helper() + cipher := testProviderTokenCipher(t) + ciphertext, version, err := cipher.Encrypt("rt-secret") + if err != nil { + t.Fatal(err) + } + store := newMemoryStreamingAccountStore() + userID := uuid.New() + if err := store.Upsert(context.Background(), StreamingAccount{ + UserID: userID, + Provider: StreamingProviderYouTube, + ChannelID: "UCabc", + RefreshTokenCiphertext: ciphertext, + TokenKeyVersion: version, + }); err != nil { + t.Fatal(err) + } + + var mu sync.Mutex + order := []string{} + hooks := map[StreamingProvider]StreamingDisconnectHooks{ + StreamingProviderYouTube: { + CleanupResources: func(context.Context, StreamingAccount) error { + mu.Lock() + order = append(order, "cleanup") + mu.Unlock() + return cleanupErr + }, + RevokeToken: func(_ context.Context, refreshToken string) error { + mu.Lock() + order = append(order, "revoke:"+refreshToken) + mu.Unlock() + return revokeErr + }, + }, + } + service, err := NewStreamingAccountService( + orderRecordingStore{StreamingAccountStore: store, mu: &mu, order: &order}, + testUserStatusChecker{status: UserStatusActive}, + cipher, + hooks, + slog.New(slog.NewTextHandler(io.Discard, nil)), + ) + if err != nil { + t.Fatal(err) + } + return service, store, userID, &order +} + +// TestDisconnectRunsStepsInOrder: ①리소스 삭제 → ②권한 취소(복호화된 RT 전달) +// → ③행 삭제 순서가 지켜져야 한다(#88 — 토큰을 먼저 폐기하면 ①이 불가능). +func TestDisconnectRunsStepsInOrder(t *testing.T) { + service, store, userID, order := disconnectFixture(t, nil, nil) + if err := service.Disconnect(context.Background(), userID, StreamingProviderYouTube); err != nil { + t.Fatal(err) + } + want := []string{"cleanup", "revoke:rt-secret", "delete"} + if len(*order) != len(want) { + t.Fatalf("order = %v, want %v", *order, want) + } + for i := range want { + if (*order)[i] != want[i] { + t.Fatalf("order = %v, want %v", *order, want) + } + } + if _, err := store.Get(context.Background(), userID, StreamingProviderYouTube); !errors.Is(err, ErrStreamingAccountNotFound) { + t.Fatal("account row must be deleted") + } +} + +// TestDisconnectContinuesWhenPlatformStepsFail: ①·②가 실패해도 ③(행 삭제)은 +// 수행돼야 한다 — 이미 토큰이 무효화된 연결을 해제하는 것이 정상 시나리오다. +func TestDisconnectContinuesWhenPlatformStepsFail(t *testing.T) { + cases := []struct { + name string + cleanupErr error + revokeErr error + }{ + {"cleanup fails", errors.New("live api down"), nil}, + {"revoke fails", nil, errors.New("revoke endpoint down")}, + {"both fail", errors.New("live api down"), errors.New("revoke endpoint down")}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + service, store, userID, order := disconnectFixture(t, tc.cleanupErr, tc.revokeErr) + if err := service.Disconnect(context.Background(), userID, StreamingProviderYouTube); err != nil { + t.Fatalf("Disconnect must succeed despite platform failures: %v", err) + } + if _, err := store.Get(context.Background(), userID, StreamingProviderYouTube); !errors.Is(err, ErrStreamingAccountNotFound) { + t.Fatal("account row must be deleted even when platform steps fail") + } + if (*order)[len(*order)-1] != "delete" { + t.Fatalf("order = %v, want delete last", *order) + } + }) + } +} + +// TestDisconnectWithoutHooksDeletesRow: 훅이 없는 플랫폼(정리할 것이 없는 +// 경우)은 행 삭제만으로 해제된다. +func TestDisconnectWithoutHooksDeletesRow(t *testing.T) { + store := newMemoryStreamingAccountStore() + userID := uuid.New() + if err := store.Upsert(context.Background(), StreamingAccount{ + UserID: userID, + Provider: StreamingProviderYouTube, + ChannelID: "UCabc", + }); err != nil { + t.Fatal(err) + } + service, err := NewStreamingAccountService(store, testUserStatusChecker{status: UserStatusActive}, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if err := service.Disconnect(context.Background(), userID, StreamingProviderYouTube); err != nil { + t.Fatal(err) + } + if _, err := store.Get(context.Background(), userID, StreamingProviderYouTube); !errors.Is(err, ErrStreamingAccountNotFound) { + t.Fatal("account row must be deleted") + } +} + +func TestDisconnectNotConnected(t *testing.T) { + service, _, _, _ := disconnectFixture(t, nil, nil) + if err := service.Disconnect(context.Background(), uuid.New(), StreamingProviderYouTube); !errors.Is(err, ErrStreamingAccountNotFound) { + t.Fatalf("error = %v, want ErrStreamingAccountNotFound", err) + } +} + +// TestRevokeTokenTreatsAlreadyInvalidAsSuccess: 이미 무효한 토큰의 revoke는 +// Google이 400을 주지만, 해제 관점에선 목적 달성이므로 에러가 아니어야 한다. +func TestRevokeTokenTreatsAlreadyInvalidAsSuccess(t *testing.T) { + var received string + okServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = r.ParseForm() + received = r.PostForm.Get("token") + w.WriteHeader(http.StatusOK) + })) + defer okServer.Close() + invalidServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + })) + defer invalidServer.Close() + brokenServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer brokenServer.Close() + + client, err := NewYouTubeOAuthClient(YouTubeOAuthConfig{ClientID: "id", ClientSecret: "secret"}) + if err != nil { + t.Fatal(err) + } + client.revokeURL = okServer.URL + if err := client.RevokeToken(context.Background(), "rt-value"); err != nil { + t.Fatal(err) + } + if received != "rt-value" { + t.Fatalf("revoked token = %q", received) + } + client.revokeURL = invalidServer.URL + if err := client.RevokeToken(context.Background(), "already-dead"); err != nil { + t.Fatalf("400 must not be an error: %v", err) + } + client.revokeURL = brokenServer.URL + if err := client.RevokeToken(context.Background(), "any"); err == nil { + t.Fatal("5xx must be an error") + } +} diff --git a/internal/auth/token_http.go b/internal/auth/token_http.go index fddf66f..d860ffc 100644 --- a/internal/auth/token_http.go +++ b/internal/auth/token_http.go @@ -94,6 +94,7 @@ func mountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLogin } if h.streamingAccounts != nil { mux.Handle("GET /auth/streaming/accounts", h.middleware(http.HandlerFunc(h.handleListStreamingAccounts))) + mux.Handle("DELETE /auth/streaming/accounts/{provider}", h.middleware(http.HandlerFunc(h.handleDisconnectStreamingAccount))) } mux.Handle("/", next) return mux diff --git a/internal/auth/youtube_oauth.go b/internal/auth/youtube_oauth.go index 2942254..0bee2dd 100644 --- a/internal/auth/youtube_oauth.go +++ b/internal/auth/youtube_oauth.go @@ -18,8 +18,9 @@ import ( ) const ( - googleOAuthTokenEndpoint = "https://oauth2.googleapis.com/token" - youtubeChannelsEndpoint = "https://www.googleapis.com/youtube/v3/channels" + googleOAuthTokenEndpoint = "https://oauth2.googleapis.com/token" + googleOAuthRevokeEndpoint = "https://oauth2.googleapis.com/revoke" + youtubeChannelsEndpoint = "https://www.googleapis.com/youtube/v3/channels" // YouTubeStreamingScope는 Live API까지 포함하는 최소 스코프다. 이보다 좁은 // 라이브 전용 스코프는 존재하지 않는다(2026-08-09 조사). 클라이언트 SDK가 // serverAuthCode를 요청할 때 같은 값을 써야 한다. @@ -131,6 +132,7 @@ type youtubeOAuthClient struct { config YouTubeOAuthConfig httpClient *http.Client tokenURL string + revokeURL string channelsURL string } @@ -142,10 +144,34 @@ func NewYouTubeOAuthClient(config YouTubeOAuthConfig) (*youtubeOAuthClient, erro config: config, httpClient: &http.Client{Timeout: 10 * time.Second}, tokenURL: googleOAuthTokenEndpoint, + revokeURL: googleOAuthRevokeEndpoint, channelsURL: youtubeChannelsEndpoint, }, nil } +// RevokeToken은 refresh token으로 부여된 권한 전체를 Google 쪽에서 취소한다 +// (연결 해제 시 사용자의 Google 계정에 "InnoLive 권한 부여됨"이 남지 않게). +// 이미 무효한 토큰이면 Google이 400을 주는데, 해제 관점에선 목적이 달성된 +// 상태이므로 에러로 다루지 않는다. +func (c *youtubeOAuthClient) RevokeToken(ctx context.Context, token string) error { + form := url.Values{"token": {strings.TrimSpace(token)}} + request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.revokeURL, strings.NewReader(form.Encode())) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/x-www-form-urlencoded") + response, err := c.httpClient.Do(request) + if err != nil { + return fmt.Errorf("request Google token revocation: %w", err) + } + defer response.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 8<<10)) + if response.StatusCode == http.StatusOK || response.StatusCode == http.StatusBadRequest { + return nil + } + return fmt.Errorf("Google token revocation returned HTTP %d", response.StatusCode) +} + // Exchange는 클라이언트가 획득한 인가 코드를 토큰으로 교환한다. 코드에 '/' // 등 예약 문자가 들어오므로(실측) 반드시 form 인코딩을 거친다. func (c *youtubeOAuthClient) Exchange(ctx context.Context, code, redirectURI string) (YouTubeTokenResponse, error) { diff --git a/internal/auth/youtube_oauth_test.go b/internal/auth/youtube_oauth_test.go index 8cab26f..7743421 100644 --- a/internal/auth/youtube_oauth_test.go +++ b/internal/auth/youtube_oauth_test.go @@ -267,6 +267,18 @@ func (s *memoryStreamingAccountStore) Upsert(_ context.Context, account Streamin return nil } +func (s *memoryStreamingAccountStore) Delete(_ context.Context, id uuid.UUID) error { + s.mu.Lock() + defer s.mu.Unlock() + for key, account := range s.accounts { + if account.ID == id { + delete(s.accounts, key) + return nil + } + } + return ErrStreamingAccountNotFound +} + func (s *memoryStreamingAccountStore) ListByUser(_ context.Context, userID uuid.UUID) ([]StreamingAccount, error) { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/streaming/youtube.go b/internal/streaming/youtube.go index f76e3d3..0efda4a 100644 --- a/internal/streaming/youtube.go +++ b/internal/streaming/youtube.go @@ -107,6 +107,36 @@ func (p *YouTubeProvider) Stop(context.Context, uuid.UUID, PreparedBroadcast) er return nil } +// CleanupStreamingResources는 연결 해제 전에 프리로딩된 재사용 스트림을 +// 플랫폼에서 삭제한다(#88 — DB 행만 지우면 사용자 채널에 고아 리소스가 +// 남고 재연결마다 누적된다). 토큰이 이미 무효면 실패하는데, 호출자 +// (StreamingAccountService)가 로그만 남기고 해제를 계속하는 계약이다. +func (p *YouTubeProvider) CleanupStreamingResources(ctx context.Context, account auth.StreamingAccount) error { + if account.StreamID == nil || *account.StreamID == "" { + return nil + } + accessToken, err := p.tokens.AccessToken(ctx, account.UserID) + if err != nil { + return fmt.Errorf("obtain access token for stream cleanup: %w", err) + } + request, err := http.NewRequestWithContext(ctx, http.MethodDelete, + p.apiBase+"/liveStreams?id="+*account.StreamID, nil) + if err != nil { + return err + } + request.Header.Set("Authorization", "Bearer "+accessToken) + response, err := p.httpClient.Do(request) + if err != nil { + return fmt.Errorf("request liveStreams.delete: %w", err) + } + defer response.Body.Close() + // 204가 정상, 404는 이미 없는 것이라 목적 달성으로 본다. + if response.StatusCode == http.StatusNoContent || response.StatusCode == http.StatusNotFound { + return nil + } + return decodeYouTubeAPIError(response) +} + // ensureReusableStream은 계정에 저장된 재사용 스트림을 복호화해 돌려주고, // 없으면 liveStreams.insert(isReusable=true)로 만들어 저장한다. 연결 시점이 // 아니라 첫 Prepare에서 lazy 생성하는 이유: 연결 서비스(auth)가 Live API에 diff --git a/internal/streaming/youtube_test.go b/internal/streaming/youtube_test.go index 51ec4e1..c99b19f 100644 --- a/internal/streaming/youtube_test.go +++ b/internal/streaming/youtube_test.go @@ -43,6 +43,16 @@ func (s *memoryStore) Upsert(_ context.Context, account auth.StreamingAccount) e return nil } +func (s *memoryStore) Delete(_ context.Context, id uuid.UUID) error { + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.accounts[id]; !ok { + return auth.ErrStreamingAccountNotFound + } + delete(s.accounts, id) + return nil +} + func (s *memoryStore) ListByUser(_ context.Context, userID uuid.UUID) ([]auth.StreamingAccount, error) { s.mu.Lock() defer s.mu.Unlock() @@ -309,6 +319,45 @@ func TestPrepareMapsLivePermissionBlocked(t *testing.T) { } } +// TestCleanupStreamingResources: 저장된 재사용 스트림이 있으면 liveStreams +// DELETE를 부르고, 없으면 API 호출 없이 무동작이어야 한다(#88). +func TestCleanupStreamingResources(t *testing.T) { + var deletes []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete || !strings.HasPrefix(r.URL.Path, "/liveStreams") { + t.Errorf("unexpected request %s %s", r.Method, r.URL.Path) + } + deletes = append(deletes, r.URL.Query().Get("id")) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + store := newMemoryStore() + provider, err := NewYouTubeProvider(stubTokens{token: "at-value"}, store, testCipher(t)) + if err != nil { + t.Fatal(err) + } + provider.apiBase = server.URL + + streamID := "stream-id-1" + withStream := auth.StreamingAccount{ID: uuid.New(), UserID: uuid.New(), Provider: auth.StreamingProviderYouTube, StreamID: &streamID} + if err := provider.CleanupStreamingResources(context.Background(), withStream); err != nil { + t.Fatal(err) + } + if len(deletes) != 1 || deletes[0] != "stream-id-1" { + t.Fatalf("deletes = %v, want [stream-id-1]", deletes) + } + + // 프리로딩된 스트림이 없는 계정은 무동작(추가 API 호출 없음). + withoutStream := auth.StreamingAccount{ID: uuid.New(), UserID: uuid.New(), Provider: auth.StreamingProviderYouTube} + if err := provider.CleanupStreamingResources(context.Background(), withoutStream); err != nil { + t.Fatal(err) + } + if len(deletes) != 1 { + t.Fatalf("deletes = %v, want no additional call", deletes) + } +} + func TestPrepareRequiresConnection(t *testing.T) { provider := testProviderWith(t, &youtubeAPIStub{}, newMemoryStore()) _, err := provider.Prepare(context.Background(), uuid.New(), PrepareOptions{}) From f7bd5e2abc3b7c4e0b7167af4f27810e4107e0dc Mon Sep 17 00:00:00 2001 From: itzjb Date: Mon, 10 Aug 2026 19:58:11 +0900 Subject: [PATCH 4/4] =?UTF-8?q?feat:=20=EB=B0=A9=EC=86=A1=20=EC=A4=80?= =?UTF-8?q?=EB=B9=84=20=EC=8B=9C=20=EC=B1=84=EB=84=90=20=ED=91=9C=EC=8B=9C?= =?UTF-8?q?=20=EC=A0=95=EB=B3=B4=20=EA=B0=B1=EC=8B=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/auth/streaming_account.go | 23 +++++++++++++ internal/auth/youtube_oauth_test.go | 14 ++++++++ internal/streaming/youtube.go | 52 +++++++++++++++++++++++++++++ internal/streaming/youtube_test.go | 37 ++++++++++++++++++++ 4 files changed, 126 insertions(+) diff --git a/internal/auth/streaming_account.go b/internal/auth/streaming_account.go index d5a1b46..da772a9 100644 --- a/internal/auth/streaming_account.go +++ b/internal/auth/streaming_account.go @@ -107,6 +107,9 @@ type StreamingAccountStore interface { MarkReconnectRequired(ctx context.Context, id uuid.UUID, at time.Time) error // Delete는 연결 행을 삭제한다. 없으면 ErrStreamingAccountNotFound. Delete(ctx context.Context, id uuid.UUID) error + // UpdateChannel은 플랫폼 쪽 채널 표시 정보를 갱신한다(사용자가 채널명을 + // 바꾼 경우의 신선도 유지 — 연결·방송 준비 시점에만 호출된다). + UpdateChannel(ctx context.Context, id uuid.UUID, channelID string, channelTitle *string) error } type gormStreamingAccountStore struct { @@ -192,6 +195,26 @@ func (s *gormStreamingAccountStore) UpdateStreamInfo(ctx context.Context, id uui }) } +func (s *gormStreamingAccountStore) UpdateChannel(ctx context.Context, id uuid.UUID, channelID string, channelTitle *string) error { + if s == nil || s.db == nil { + return errors.New("streaming account database is nil") + } + result := s.db.WithContext(ctx).Model(&StreamingAccount{}). + Where("id = ?", id). + Updates(map[string]any{ + "channel_id": channelID, + "channel_title": channelTitle, + "updated_at": s.now(), + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected == 0 { + return ErrStreamingAccountNotFound + } + return nil +} + func (s *gormStreamingAccountStore) Delete(ctx context.Context, id uuid.UUID) error { if s == nil || s.db == nil { return errors.New("streaming account database is nil") diff --git a/internal/auth/youtube_oauth_test.go b/internal/auth/youtube_oauth_test.go index 7743421..77d608c 100644 --- a/internal/auth/youtube_oauth_test.go +++ b/internal/auth/youtube_oauth_test.go @@ -267,6 +267,20 @@ func (s *memoryStreamingAccountStore) Upsert(_ context.Context, account Streamin return nil } +func (s *memoryStreamingAccountStore) UpdateChannel(_ context.Context, id uuid.UUID, channelID string, channelTitle *string) error { + s.mu.Lock() + defer s.mu.Unlock() + for key, account := range s.accounts { + if account.ID == id { + account.ChannelID = channelID + account.ChannelTitle = channelTitle + s.accounts[key] = account + return nil + } + } + return ErrStreamingAccountNotFound +} + func (s *memoryStreamingAccountStore) Delete(_ context.Context, id uuid.UUID) error { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/streaming/youtube.go b/internal/streaming/youtube.go index 0efda4a..68eaa5c 100644 --- a/internal/streaming/youtube.go +++ b/internal/streaming/youtube.go @@ -85,6 +85,10 @@ func (p *YouTubeProvider) Prepare(ctx context.Context, userID uuid.UUID, options if err != nil { return PreparedBroadcast{}, err } + // 방송 준비 시점의 채널 표시 정보 갱신(#88 ④) — 조회 API가 저장값을 + // 반환하는 대가로 여기서 신선도를 맞춘다(1 unit). 부가 기능이므로 실패해도 + // 방송 준비는 계속한다. + p.refreshChannelInfo(ctx, accessToken, account) broadcastID, err := p.insertBroadcast(ctx, accessToken, options) if err != nil { return PreparedBroadcast{}, err @@ -195,6 +199,54 @@ func (p *YouTubeProvider) ensureReusableStream(ctx context.Context, accessToken return response.ID, info.RtmpsIngestionAddress, info.StreamName, nil } +// refreshChannelInfo는 channels.list(mine=true)로 현재 채널 정보를 조회해 +// 저장값과 다르면 갱신한다. 실패는 로그 대상도 아닌 무시다 — 표시 정보의 +// 신선도일 뿐 방송 준비의 성패와 무관하다. +func (p *YouTubeProvider) refreshChannelInfo(ctx context.Context, accessToken string, account auth.StreamingAccount) { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, p.apiBase+"/channels?part=snippet&mine=true", nil) + if err != nil { + return + } + request.Header.Set("Authorization", "Bearer "+accessToken) + response, err := p.httpClient.Do(request) + if err != nil { + return + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 8<<10)) + return + } + var payload struct { + Items []struct { + ID string `json:"id"` + Snippet struct { + Title string `json:"title"` + } `json:"snippet"` + } `json:"items"` + } + if err := json.NewDecoder(io.LimitReader(response.Body, 256<<10)).Decode(&payload); err != nil { + return + } + if len(payload.Items) == 0 || payload.Items[0].ID == "" { + return + } + channelID := payload.Items[0].ID + title := strings.TrimSpace(payload.Items[0].Snippet.Title) + storedTitle := "" + if account.ChannelTitle != nil { + storedTitle = *account.ChannelTitle + } + if channelID == account.ChannelID && title == storedTitle { + return + } + var titlePtr *string + if title != "" { + titlePtr = &title + } + _ = p.store.UpdateChannel(ctx, account.ID, channelID, titlePtr) +} + func (p *YouTubeProvider) insertBroadcast(ctx context.Context, accessToken string, options PrepareOptions) (string, error) { title := strings.TrimSpace(options.Title) if title == "" { diff --git a/internal/streaming/youtube_test.go b/internal/streaming/youtube_test.go index c99b19f..5251be8 100644 --- a/internal/streaming/youtube_test.go +++ b/internal/streaming/youtube_test.go @@ -43,6 +43,19 @@ func (s *memoryStore) Upsert(_ context.Context, account auth.StreamingAccount) e return nil } +func (s *memoryStore) UpdateChannel(_ context.Context, id uuid.UUID, channelID string, channelTitle *string) error { + s.mu.Lock() + defer s.mu.Unlock() + account, ok := s.accounts[id] + if !ok { + return auth.ErrStreamingAccountNotFound + } + account.ChannelID = channelID + account.ChannelTitle = channelTitle + s.accounts[id] = account + return nil +} + func (s *memoryStore) Delete(_ context.Context, id uuid.UUID) error { s.mu.Lock() defer s.mu.Unlock() @@ -146,6 +159,9 @@ func (s *youtubeAPIStub) handler(t *testing.T) http.Handler { s.mu.Lock() defer s.mu.Unlock() switch { + case strings.HasPrefix(r.URL.Path, "/channels"): + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"items":[{"id":"UCabc","snippet":{"title":"Team Framework Renamed"}}]}`)) case strings.HasPrefix(r.URL.Path, "/liveStreams"): s.streamInserts++ w.Header().Set("Content-Type", "application/json") @@ -319,6 +335,27 @@ func TestPrepareMapsLivePermissionBlocked(t *testing.T) { } } +// TestPrepareRefreshesChannelInfo: 방송 준비가 채널 표시 정보를 갱신해야 +// 한다(#88 ④ — 조회 API는 저장값을 반환하므로 여기서 신선도를 맞춘다). +func TestPrepareRefreshesChannelInfo(t *testing.T) { + stub := &youtubeAPIStub{} + store := newMemoryStore() + userID := uuid.New() + connectedAccount(t, store, userID) // 저장된 제목은 없음(nil), 스텁은 "Team Framework Renamed"를 반환 + provider := testProviderWith(t, stub, store) + + if _, err := provider.Prepare(context.Background(), userID, PrepareOptions{}); err != nil { + t.Fatal(err) + } + account, err := store.Get(context.Background(), userID, auth.StreamingProviderYouTube) + if err != nil { + t.Fatal(err) + } + if account.ChannelTitle == nil || *account.ChannelTitle != "Team Framework Renamed" { + t.Fatalf("channel title = %v, want refreshed value", account.ChannelTitle) + } +} + // TestCleanupStreamingResources: 저장된 재사용 스트림이 있으면 liveStreams // DELETE를 부르고, 없으면 API 호출 없이 무동작이어야 한다(#88). func TestCleanupStreamingResources(t *testing.T) {