feat: streaming-connection-api/#88

cmd/server/main.go

@@ -320,6 +320,10 @@ func main() {
 	// 배포에서도 refresh token 암호화가 성립해야 한다.
 	var youtubeConnect *auth.YouTubeConnectService
 	streamingProviders := map[auth.StreamingProvider]streaming.Provider{}
+	// 송출 계정 저장소·조회 서비스는 플랫폼 중립이라 YouTube 설정 여부와
+	// 무관하게 조립한다 — 연결이 없으면 조회가 빈 배열을 돌려줄 뿐이다.
+	streamingAccountStore := auth.NewGormStreamingAccountStore(databaseConnection.DB)
+	streamingDisconnectHooks := map[auth.StreamingProvider]auth.StreamingDisconnectHooks{}
 	if youtubeOAuthConfig.Enabled() {
 		if providerTokenCipher == nil {
 			providerTokenCipher, err = auth.NewProviderTokenCipherFromBase64(os.Getenv("AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64"))
@@ -333,7 +337,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,
@@ -355,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
@@ -422,7 +436,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,
 	}

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"`
@@ -89,12 +95,21 @@ 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을 담아온 경우
 	// 행 락 하에 교체한다 — 한 사용자의 다중 세션이 동시에 갱신할 때 나중에
 	// 실패한 쓰기가 최신 토큰을 덮지 않도록 잠근다.
 	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
+	// 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 {
@@ -123,9 +138,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 +195,73 @@ 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")
+	}
+	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")
+	}
+	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")
+	}
+	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")

internal/auth/streaming_account_http.go

@@ -0,0 +1,87 @@
+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)
+}
+
+// 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
+}