feat: youtube-oauth-egress/#83

.env.example

@@ -43,8 +43,9 @@ WEBRTC_UDP_PORT_MIN=50002
 WEBRTC_UDP_PORT_MAX=50020
 WEBRTC_DISCONNECTED_GRACE=10s
 
-# ── YouTube RTMP egress (발행자 오디오) ────────────────
-YOUTUBE_STREAM_KEY=
+# ── RTMP egress (발행자 오디오) ────────────────
+# 송출 대상은 이제 서버 전역 키가 아니라 사용자별 연결 계정(/auth/youtube/connect)에서
+# 나오며, 송출 개시는 POST /sessions/{id}/stream/start 로만 이루어집니다.
 ENABLE_AUDIO_EGRESS=false            # 발행자 마이크(Opus)를 함께 송출. off면 무음
 EGRESS_LATENCY_LOG=false             # ingest→egress 지연 p50/p95/max 로깅(측정용)
 EGRESS_AUDIO_OFFSET_MS=0             # A/V 싱크 보정(오디오 -itsoffset, ms). 양수=오디오 지연
@@ -93,6 +94,15 @@ APPLE_PRIVATE_KEY_PATH=
 # openssl rand -base64 32 로 생성한 32바이트 AES-256 키입니다. 절대로 저장소에 커밋하지 마세요.
 AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64=
 
+# YouTube 송출 연동 (로그인용 Google OAuth와 별개의 전용 GCP 프로젝트/클라이언트)
+# CLIENT_ID/SECRET 둘 다 설정해야 활성화되며, 활성 시 AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64도 필요합니다.
+# 클라이언트(네이티브 GoogleSignIn SDK 또는 웹 GIS 팝업)가 획득한 인가 코드를
+# POST /auth/youtube/connect {server_auth_code, code_source} 로 받습니다. redirect URI 설정은 없습니다 —
+# 교환 파라미터는 code_source(native|web_popup)별로 서버가 결정합니다.
+# 웹 팝업을 쓰려면 GCP 콘솔의 웹 클라이언트에 "승인된 JavaScript 원본"(예: http://localhost:8000)이 등록돼야 합니다.
+YOUTUBE_OAUTH_CLIENT_ID=
+YOUTUBE_OAUTH_CLIENT_SECRET=
+
 # Email/password authentication (SMTP + Redis)
 # AUTH_EMAIL_SMTP_HOST가 비어 있으면 이메일 회원가입 API는 비활성화됩니다.
 # 587 포트는 STARTTLS=true, 465 포트는 IMPLICIT_TLS=true 및 STARTTLS=false가 일반적입니다.

cmd/server/main.go

@@ -23,6 +23,7 @@ import (
 	"inno-live-server/internal/origin"
 	"inno-live-server/internal/server"
 	"inno-live-server/internal/session"
+	"inno-live-server/internal/streaming"
 )
 
 func main() {
@@ -295,6 +296,11 @@ func main() {
 		logger.Error("create account withdrawal service failed", "error", err)
 		os.Exit(2)
 	}
+	youtubeOAuthConfig, err := auth.LoadYouTubeOAuthConfigFromEnv()
+	if err != nil {
+		logger.Error("invalid YouTube OAuth configuration", "error", err)
+		os.Exit(2)
+	}
 	logger.Info(
 		"authentication token service ready",
 		"access_ttl", tokenConfig.AccessTTL,
@@ -305,9 +311,51 @@ func main() {
 		"google_oauth_enabled", googleOAuthConfig.Enabled(),
 		"apple_oauth_enabled", appleOAuthConfig.Enabled(),
 		"email_auth_enabled", emailAuthConfig.Enabled(),
+		"youtube_streaming_enabled", youtubeOAuthConfig.Enabled(),
 	)
 
 	userStatusChecker := auth.NewGormUserStatusChecker(databaseConnection.DB)
+	// YouTube 송출 연동. 암호화 키(cipher)는 Apple 활성 시 위에서 이미
+	// 생성됐을 수 있으므로 없을 때만 만든다 — Apple 없이 YouTube만 켠
+	// 배포에서도 refresh token 암호화가 성립해야 한다.
+	var youtubeConnect *auth.YouTubeConnectService
+	streamingProviders := map[auth.StreamingProvider]streaming.Provider{}
+	if youtubeOAuthConfig.Enabled() {
+		if providerTokenCipher == nil {
+			providerTokenCipher, err = auth.NewProviderTokenCipherFromBase64(os.Getenv("AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64"))
+			if err != nil {
+				logger.Error("invalid provider token encryption configuration", "error", err)
+				os.Exit(2)
+			}
+		}
+		youtubeOAuthClient, err := auth.NewYouTubeOAuthClient(youtubeOAuthConfig)
+		if err != nil {
+			logger.Error("create YouTube OAuth client failed", "error", err)
+			os.Exit(2)
+		}
+		streamingAccountStore := auth.NewGormStreamingAccountStore(databaseConnection.DB)
+		youtubeConnect, err = auth.NewYouTubeConnectService(
+			youtubeOAuthClient,
+			streamingAccountStore,
+			userStatusChecker,
+			providerTokenCipher,
+		)
+		if err != nil {
+			logger.Error("create YouTube connect service failed", "error", err)
+			os.Exit(2)
+		}
+		youtubeTokens, err := auth.NewYouTubeAccessTokenProvider(youtubeOAuthClient, streamingAccountStore, providerTokenCipher)
+		if err != nil {
+			logger.Error("create YouTube access token provider failed", "error", err)
+			os.Exit(2)
+		}
+		youtubeProvider, err := streaming.NewYouTubeProvider(youtubeTokens, streamingAccountStore, providerTokenCipher)
+		if err != nil {
+			logger.Error("create YouTube streaming provider failed", "error", err)
+			os.Exit(2)
+		}
+		streamingProviders[auth.StreamingProviderYouTube] = youtubeProvider
+	}
 	// INNOLIVE_REQUIRE_SESSION_AUTH=false is the explicit local-development
 	// escape hatch (loud warning above). Extend it to user auth as well so
 	// tokenless bench tooling (pion-load) keeps working against dev servers;
@@ -328,6 +376,7 @@ func main() {
 		aiPool,
 		originConfig,
 		requireUser,
+		streamingProviders,
 		authenticateUser,
 	)
 
@@ -373,7 +422,7 @@ func main() {
 
 	httpServer := &http.Server{
 		Addr:              cfg.HTTPAddr,
-		Handler:           auth.MountAuthHTTPWithServices(application.Handler(), tokenService, googleLogin, appleLogin, emailLogin, withdrawal, logger, originConfig),
+		Handler:           auth.MountAuthHTTPWithServices(application.Handler(), tokenService, googleLogin, appleLogin, emailLogin, withdrawal, logger, originConfig, youtubeConnect),
 		ReadHeaderTimeout: 10 * time.Second,
 		IdleTimeout:       60 * time.Second,
 	}

internal/auth/migrate.go

@@ -18,6 +18,7 @@ func AutoMigrate(ctx context.Context, db *gorm.DB) error {
 		&OAuthAccount{},
 		&EmailAccount{},
 		&RefreshSession{},
+		&StreamingAccount{},
 	); err != nil {
 		return fmt.Errorf("auto migrate authentication schema: %w", err)
 	}

internal/auth/provider_token_cipher.go

@@ -23,7 +23,7 @@ type ProviderTokenCipher struct {
 func NewProviderTokenCipherFromBase64(value string) (*ProviderTokenCipher, error) {
 	value = strings.TrimSpace(value)
 	if value == "" {
-		return nil, errors.New("AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64 is required when Apple OAuth is enabled")
+		return nil, errors.New("AUTH_PROVIDER_TOKEN_ENCRYPTION_KEY_BASE64 is required when Apple OAuth or a streaming integration is enabled")
 	}
 	key, err := base64.StdEncoding.DecodeString(value)
 	if err != nil {

internal/auth/streaming_account.go

@@ -0,0 +1,204 @@
+package auth
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+// StreamingProvider는 RTMP 송출 대상 플랫폼 식별자다. 로그인 공급자
+// (OAuthProvider)와는 별개 개념이다 — 로그인은 이메일로 하고 송출은
+// YouTube 채널로 하는 식으로 연결이 독립적이다.
+type StreamingProvider string
+
+const (
+	StreamingProviderYouTube StreamingProvider = "youtube"
+	StreamingProviderChzzk   StreamingProvider = "chzzk"
+)
+
+var ErrStreamingAccountNotFound = errors.New("streaming account not found")
+
+// StreamingAccount는 사용자가 연결한 송출 플랫폼 계정이다. 사용자당 플랫폼별
+// 1개(멀티 채널 불허)로 시작하며, 멀티 채널이 필요해지면 유니크 인덱스를 풀고
+// 선택 채널 포인터를 추가하는 마이그레이션으로 확장한다.
+type StreamingAccount struct {
+	ID uuid.UUID `gorm:"type:uuid;primaryKey"`
+
+	UserID uuid.UUID `gorm:"type:uuid;not null;index;uniqueIndex:uidx_streaming_user_provider,priority:1"`
+	User   *User     `gorm:"foreignKey:UserID;references:ID;constraint:OnUpdate:CASCADE,OnDelete:CASCADE"`
+
+	Provider StreamingProvider `gorm:"type:varchar(20);not null;uniqueIndex:uidx_streaming_user_provider,priority:2;check:chk_streaming_provider,provider IN ('youtube','chzzk')"`
+
+	// 연결 상태 표시("○○ 채널에 연결됨")에 쓰는 플랫폼 쪽 채널 식별 정보.
+	// 연결(콜백) 시점에 플랫폼 API로 조회해 저장한다.
+	ChannelID    string  `gorm:"type:varchar(255);not null"`
+	ChannelTitle *string `gorm:"type:varchar(255)"`
+
+	// 플랫폼 OAuth refresh token의 AES-GCM 암호문. 평문은 저장하지 않는다.
+	RefreshTokenCiphertext []byte `gorm:"type:bytea"`
+	TokenKeyVersion        *int16 `gorm:"type:smallint"`
+
+	// refresh token 자체의 만료 시각. Google OAuth는 앱 게시 상태가 Testing이면
+	// 토큰 응답에 refresh_token_expires_in(실측 7일)을 담아 보낸다 — 무기한이면
+	// NULL이다. 이 값을 추적하지 않으면 연결이 만료 후 조용히 죽는다.
+	RefreshTokenExpiresAt *time.Time
+
+	// 치지직처럼 ingest URL을 API로 제공하지 않는 플랫폼을 위한 수동 설정값.
+	// 연결(OAuth) 플로우는 이 컬럼을 건드리지 않는다.
+	ManualIngestURL *string `gorm:"type:text"`
+
+	// 이하 YouTube 재사용 스트림(isReusable=true) 프리로딩 정보 — 첫 방송
+	// 준비 때 1회 생성해 저장하고 이후 방송은 재사용한다(방송당 API 3→2회).
+	// 필드 구성은 liveStreams.insert 응답의 cdn.ingestionInfo 실물
+	// (2026-08-10 실측: rtmp/rtmps × 주/백업 4주소 + streamName)을 따른다.
+	StreamID                    *string `gorm:"type:varchar(255)"`
+	IngestionAddress            *string `gorm:"type:text"`
+	BackupIngestionAddress      *string `gorm:"type:text"`
+	RtmpsIngestionAddress       *string `gorm:"type:text"`
+	RtmpsBackupIngestionAddress *string `gorm:"type:text"`
+	// streamName은 RTMP URL에 붙는 스트림 키 상당의 비밀값이라 refresh token과
+	// 같은 방식(AES-GCM)으로 암호화해서만 저장한다.
+	StreamNameCiphertext []byte `gorm:"type:bytea"`
+	StreamNameKeyVersion *int16 `gorm:"type:smallint"`
+
+	ConnectedAt time.Time `gorm:"not null"`
+	CreatedAt   time.Time `gorm:"not null"`
+	UpdatedAt   time.Time `gorm:"not null"`
+}
+
+// StreamInfo는 프리로딩된 재사용 스트림 정보의 갱신 단위다.
+type StreamInfo struct {
+	StreamID                    string
+	IngestionAddress            string
+	BackupIngestionAddress      string
+	RtmpsIngestionAddress       string
+	RtmpsBackupIngestionAddress string
+	StreamNameCiphertext        []byte
+	StreamNameKeyVersion        *int16
+}
+
+func (StreamingAccount) TableName() string { return "streaming_accounts" }
+
+// StreamingAccountStore는 송출 계정 연결의 영속화 계약이다.
+type StreamingAccountStore interface {
+	// Upsert는 연결을 저장한다. 같은 (user, provider) 재연결이면 채널 정보와
+	// 토큰을 교체하고 기존 행(ID)을 유지한다.
+	Upsert(ctx context.Context, account StreamingAccount) error
+	Get(ctx context.Context, userID uuid.UUID, provider StreamingProvider) (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
+}
+
+type gormStreamingAccountStore struct {
+	db  *gorm.DB
+	now func() time.Time
+}
+
+func NewGormStreamingAccountStore(db *gorm.DB) StreamingAccountStore {
+	return &gormStreamingAccountStore{db: db, now: func() time.Time { return time.Now().UTC() }}
+}
+
+func (s *gormStreamingAccountStore) Upsert(ctx context.Context, account StreamingAccount) error {
+	if s == nil || s.db == nil {
+		return errors.New("streaming account database is nil")
+	}
+	now := s.now()
+	if account.ID == uuid.Nil {
+		account.ID = uuid.New()
+	}
+	account.ConnectedAt = now
+	account.CreatedAt = now
+	account.UpdatedAt = now
+	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := ensureActiveUser(tx, account.UserID); err != nil {
+			return err
+		}
+		return tx.Clauses(clause.OnConflict{
+			Columns: []clause.Column{{Name: "user_id"}, {Name: "provider"}},
+			DoUpdates: clause.AssignmentColumns([]string{
+				"channel_id", "channel_title",
+				"refresh_token_ciphertext", "token_key_version", "refresh_token_expires_at",
+				"connected_at", "updated_at",
+			}),
+		}).Create(&account).Error
+	})
+}
+
+func (s *gormStreamingAccountStore) Get(ctx context.Context, userID uuid.UUID, provider StreamingProvider) (StreamingAccount, error) {
+	if s == nil || s.db == nil {
+		return StreamingAccount{}, errors.New("streaming account database is nil")
+	}
+	var account StreamingAccount
+	result := s.db.WithContext(ctx).
+		Where("user_id = ? AND provider = ?", userID, provider).
+		Take(&account)
+	if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+		return StreamingAccount{}, ErrStreamingAccountNotFound
+	}
+	if result.Error != nil {
+		return StreamingAccount{}, result.Error
+	}
+	return account, nil
+}
+
+func (s *gormStreamingAccountStore) UpdateStreamInfo(ctx context.Context, id uuid.UUID, info StreamInfo) error {
+	if s == nil || s.db == nil {
+		return errors.New("streaming account database is nil")
+	}
+	now := s.now()
+	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var current StreamingAccount
+		result := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+			Where("id = ?", id).
+			Take(&current)
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return ErrStreamingAccountNotFound
+		}
+		if result.Error != nil {
+			return result.Error
+		}
+		return tx.Model(&StreamingAccount{}).Where("id = ?", id).Updates(map[string]any{
+			"stream_id":                      info.StreamID,
+			"ingestion_address":              info.IngestionAddress,
+			"backup_ingestion_address":       info.BackupIngestionAddress,
+			"rtmps_ingestion_address":        info.RtmpsIngestionAddress,
+			"rtmps_backup_ingestion_address": info.RtmpsBackupIngestionAddress,
+			"stream_name_ciphertext":         info.StreamNameCiphertext,
+			"stream_name_key_version":        info.StreamNameKeyVersion,
+			"updated_at":                     now,
+		}).Error
+	})
+}
+
+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")
+	}
+	now := s.now()
+	return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var current StreamingAccount
+		result := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+			Where("id = ?", id).
+			Take(&current)
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return ErrStreamingAccountNotFound
+		}
+		if result.Error != nil {
+			return result.Error
+		}
+		return tx.Model(&StreamingAccount{}).Where("id = ?", id).Updates(map[string]any{
+			"refresh_token_ciphertext": ciphertext,
+			"token_key_version":        version,
+			"refresh_token_expires_at": expiresAt,
+			"updated_at":               now,
+		}).Error
+	})
+}