diff --git a/.env.example b/.env.example
index b092f12..08a0b31 100644
--- a/.env.example
+++ b/.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가 일반적입니다.
diff --git a/cmd/server/main.go b/cmd/server/main.go
index 14284ca..630504e 100644
--- a/cmd/server/main.go
+++ b/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,
}
diff --git a/internal/auth/migrate.go b/internal/auth/migrate.go
index aeaa677..1b091e8 100644
--- a/internal/auth/migrate.go
+++ b/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)
}
diff --git a/internal/auth/provider_token_cipher.go b/internal/auth/provider_token_cipher.go
index f9f5984..d292864 100644
--- a/internal/auth/provider_token_cipher.go
+++ b/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 {
diff --git a/internal/auth/streaming_account.go b/internal/auth/streaming_account.go
new file mode 100644
index 0000000..f56a209
--- /dev/null
+++ b/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(¤t)
+ 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(¤t)
+ 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
+ })
+}
diff --git a/internal/auth/streaming_account_postgres_test.go b/internal/auth/streaming_account_postgres_test.go
new file mode 100644
index 0000000..bf12edb
--- /dev/null
+++ b/internal/auth/streaming_account_postgres_test.go
@@ -0,0 +1,138 @@
+package auth
+
+import (
+ "context"
+ "errors"
+ "net/url"
+ "os"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+ "gorm.io/driver/postgres"
+ "gorm.io/gorm"
+)
+
+// TestPostgresStreamingAccountUpsert verifies the PostgreSQL-specific pieces a
+// memory double cannot: the ON CONFLICT (user_id, provider) upsert path, the
+// unique index, and the row-locked refresh-token update. Set TEST_DATABASE_URL
+// to run it; each run uses an isolated, temporary schema.
+func TestPostgresStreamingAccountUpsert(t *testing.T) {
+ databaseURL := strings.TrimSpace(os.Getenv("TEST_DATABASE_URL"))
+ if databaseURL == "" {
+ t.Skip("set TEST_DATABASE_URL to run PostgreSQL streaming account integration test")
+ }
+
+ db := newPostgresStreamingTestDB(t, databaseURL)
+ now := time.Now().UTC().Truncate(time.Microsecond)
+ user := User{ID: uuid.New(), Status: UserStatusActive, CreatedAt: now, UpdatedAt: now}
+ if err := db.Create(&user).Error; err != nil {
+ t.Fatal(err)
+ }
+ store := NewGormStreamingAccountStore(db)
+ ctx := context.Background()
+
+ title := "Team Framework"
+ first := StreamingAccount{
+ UserID: user.ID,
+ Provider: StreamingProviderYouTube,
+ ChannelID: "UCfirst",
+ ChannelTitle: &title,
+ RefreshTokenCiphertext: []byte{1, 2, 3},
+ }
+ if err := store.Upsert(ctx, first); err != nil {
+ t.Fatal(err)
+ }
+ created, err := store.Get(ctx, user.ID, StreamingProviderYouTube)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // 재연결: 같은 (user, provider)는 새 행이 아니라 기존 행 갱신이어야 한다.
+ second := StreamingAccount{
+ UserID: user.ID,
+ Provider: StreamingProviderYouTube,
+ ChannelID: "UCsecond",
+ RefreshTokenCiphertext: []byte{4, 5, 6},
+ }
+ if err := store.Upsert(ctx, second); err != nil {
+ t.Fatal(err)
+ }
+ updated, err := store.Get(ctx, user.ID, StreamingProviderYouTube)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if updated.ID != created.ID {
+ t.Fatalf("re-connect created a new row: %s -> %s", created.ID, updated.ID)
+ }
+ if updated.ChannelID != "UCsecond" || string(updated.RefreshTokenCiphertext) != string([]byte{4, 5, 6}) {
+ t.Fatalf("re-connect did not replace channel/token: %+v", updated)
+ }
+ var count int64
+ if err := db.Model(&StreamingAccount{}).Where("user_id = ?", user.ID).Count(&count).Error; err != nil {
+ t.Fatal(err)
+ }
+ if count != 1 {
+ t.Fatalf("rows for user = %d, want 1 (unique index)", count)
+ }
+
+ // 행 락 기반 refresh token 교체.
+ expiresAt := now.Add(7 * 24 * time.Hour)
+ version := int16(1)
+ if err := store.UpdateRefreshToken(ctx, updated.ID, []byte{7, 8, 9}, &version, &expiresAt); err != nil {
+ t.Fatal(err)
+ }
+ rotated, err := store.Get(ctx, user.ID, StreamingProviderYouTube)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(rotated.RefreshTokenCiphertext) != string([]byte{7, 8, 9}) || rotated.RefreshTokenExpiresAt == nil {
+ t.Fatalf("refresh token not rotated: %+v", rotated)
+ }
+ if err := store.UpdateRefreshToken(ctx, uuid.New(), []byte{0}, &version, nil); !errors.Is(err, ErrStreamingAccountNotFound) {
+ t.Fatalf("unknown id error = %v, want ErrStreamingAccountNotFound", err)
+ }
+
+ // 비활성 사용자의 연결은 거부돼야 한다.
+ disabled := User{ID: uuid.New(), Status: UserStatusDisabled, CreatedAt: now, UpdatedAt: now}
+ if err := db.Create(&disabled).Error; err != nil {
+ t.Fatal(err)
+ }
+ if err := store.Upsert(ctx, StreamingAccount{UserID: disabled.ID, Provider: StreamingProviderYouTube, ChannelID: "UCx"}); !errors.Is(err, ErrUserInactive) {
+ t.Fatalf("disabled user upsert error = %v, want ErrUserInactive", err)
+ }
+}
+
+func newPostgresStreamingTestDB(t *testing.T, databaseURL string) *gorm.DB {
+ t.Helper()
+ admin, err := gorm.Open(postgres.Open(databaseURL), &gorm.Config{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ schema := "auth_streaming_test_" + strings.ReplaceAll(uuid.NewString(), "-", "")
+ if err := admin.Exec("CREATE SCHEMA " + schema).Error; err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() {
+ if err := admin.Exec("DROP SCHEMA " + schema + " CASCADE").Error; err != nil {
+ t.Errorf("drop PostgreSQL test schema: %v", err)
+ }
+ })
+
+ parsed, err := url.Parse(databaseURL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ query := parsed.Query()
+ query.Set("search_path", schema)
+ parsed.RawQuery = query.Encode()
+ db, err := gorm.Open(postgres.Open(parsed.String()), &gorm.Config{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := db.AutoMigrate(&User{}, &StreamingAccount{}); err != nil {
+ t.Fatal(err)
+ }
+ return db
+}
diff --git a/internal/auth/token_http.go b/internal/auth/token_http.go
index 90e1028..686517d 100644
--- a/internal/auth/token_http.go
+++ b/internal/auth/token_http.go
@@ -22,6 +22,7 @@ type tokenHTTPHandler struct {
apple *AppleLoginService
email *EmailAuthService
withdrawal *AccountWithdrawalService
+ youtube *YouTubeConnectService
logger *slog.Logger
config TokenHTTPConfig
}
@@ -35,26 +36,32 @@ func MountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLogin
if len(appleServices) > 0 {
apple = appleServices[0]
}
- return mountAuthHTTP(next, service, google, apple, nil, nil, logger, config)
+ return mountAuthHTTP(next, service, google, apple, 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, logger, config)
+ return mountAuthHTTP(next, service, google, apple, nil, withdrawal, nil, logger, config)
}
// MountAuthHTTPWithServices mounts all configured authentication services.
-func MountAuthHTTPWithServices(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, email *EmailAuthService, withdrawal *AccountWithdrawalService, logger *slog.Logger, config TokenHTTPConfig) http.Handler {
- return mountAuthHTTP(next, service, google, apple, email, withdrawal, logger, config)
+// youtubeServices는 송출 연동(YouTube OAuth)을 함께 마운트할 때만 전달한다 —
+// 기존 호출부(테스트 포함)를 깨지 않도록 가변 인자로 확장했다.
+func MountAuthHTTPWithServices(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, email *EmailAuthService, withdrawal *AccountWithdrawalService, logger *slog.Logger, config TokenHTTPConfig, youtubeServices ...*YouTubeConnectService) http.Handler {
+ var youtube *YouTubeConnectService
+ if len(youtubeServices) > 0 {
+ youtube = youtubeServices[0]
+ }
+ return mountAuthHTTP(next, service, google, apple, email, withdrawal, youtube, logger, config)
}
-func mountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, email *EmailAuthService, withdrawal *AccountWithdrawalService, logger *slog.Logger, config TokenHTTPConfig) http.Handler {
+func mountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLoginService, apple *AppleLoginService, email *EmailAuthService, withdrawal *AccountWithdrawalService, youtube *YouTubeConnectService, 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, logger: logger, config: config}
+ h := &tokenHTTPHandler{service: service, google: google, apple: apple, email: email, withdrawal: withdrawal, youtube: youtube, 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)))
@@ -74,6 +81,10 @@ func mountAuthHTTP(next http.Handler, service *TokenService, google *GoogleLogin
if h.withdrawal != nil {
mux.Handle("DELETE /auth/me", h.middleware(http.HandlerFunc(h.handleWithdrawal)))
}
+ if h.youtube != nil {
+ mux.Handle("POST /auth/youtube/connect", h.middleware(http.HandlerFunc(h.handleYouTubeConnect)))
+ mux.Handle("GET /auth/youtube/config", h.middleware(http.HandlerFunc(h.handleYouTubeConfig)))
+ }
mux.Handle("/", next)
return mux
}
diff --git a/internal/auth/youtube_oauth.go b/internal/auth/youtube_oauth.go
new file mode 100644
index 0000000..ce06bf3
--- /dev/null
+++ b/internal/auth/youtube_oauth.go
@@ -0,0 +1,421 @@
+package auth
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "os"
+ "strings"
+ "sync"
+ "time"
+ "unicode/utf8"
+
+ "github.com/google/uuid"
+)
+
+const (
+ googleOAuthTokenEndpoint = "https://oauth2.googleapis.com/token"
+ youtubeChannelsEndpoint = "https://www.googleapis.com/youtube/v3/channels"
+ // YouTubeStreamingScope는 Live API까지 포함하는 최소 스코프다. 이보다 좁은
+ // 라이브 전용 스코프는 존재하지 않는다(2026-08-09 조사). 클라이언트 SDK가
+ // serverAuthCode를 요청할 때 같은 값을 써야 한다.
+ YouTubeStreamingScope = "https://www.googleapis.com/auth/youtube"
+ maxYouTubeOAuthField = 512
+ // accessTokenExpirySlack만큼 만료를 앞당겨 취급해, 발급 직후 만료되는
+ // 토큰으로 API를 치는 경계 상황을 피한다.
+ accessTokenExpirySlack = 60 * time.Second
+)
+
+var (
+ ErrYouTubeChannelMissing = errors.New("Google account has no YouTube channel")
+ ErrYouTubeTokenExchange = errors.New("YouTube token exchange failed")
+ ErrYouTubeAuthCodeRejected = errors.New("YouTube authorization code was rejected")
+ ErrStreamingNotConnected = errors.New("streaming account is not connected")
+)
+
+// CodeSource는 인가 코드를 발급받은 클라이언트 유형이다. 교환 시 요구되는
+// redirect_uri가 유형마다 달라(2026-08-10 실측: 웹 팝업 코드는 생략 시
+// "Missing parameter: redirect_uri"로 거절, redirect_uri=postmessage 필수)
+// 클라이언트가 출처를 선언하고 서버가 매핑한다. 출처를 잘못 선언해도 교환이
+// 실패할 뿐 보안 영향은 없다(교환 자격증명은 서버의 웹 클라이언트 것 하나).
+type CodeSource string
+
+const (
+ // CodeSourceNative: GoogleSignIn SDK의 serverAuthCode. 교환 시
+ // redirect_uri 생략(문서 기준 — iOS 실물 코드로 최종 확인 예정).
+ CodeSourceNative CodeSource = "native"
+ // CodeSourceWebPopup: GIS initCodeClient(ux_mode: popup)가 발급한 코드.
+ // 교환 시 redirect_uri=postmessage 필수(실측 확정).
+ CodeSourceWebPopup CodeSource = "web_popup"
+)
+
+func (s CodeSource) Valid() bool {
+ return s == CodeSourceNative || s == CodeSourceWebPopup
+}
+
+// exchangeRedirectURI는 출처별 교환 redirect_uri 값이다. 빈 문자열은
+// 파라미터 생략을 뜻한다.
+func (s CodeSource) exchangeRedirectURI() string {
+ if s == CodeSourceWebPopup {
+ return "postmessage"
+ }
+ return ""
+}
+
+// YouTubeOAuthConfig는 송출 연동 전용 Google OAuth 클라이언트 설정이다.
+// 로그인용(GOOGLE_OAUTH_WEB_CLIENT_ID)과는 별도 GCP 프로젝트·클라이언트다.
+type YouTubeOAuthConfig struct {
+ ClientID string
+ ClientSecret string
+}
+
+func LoadYouTubeOAuthConfigFromEnv() (YouTubeOAuthConfig, error) {
+ config := YouTubeOAuthConfig{
+ ClientID: strings.TrimSpace(os.Getenv("YOUTUBE_OAUTH_CLIENT_ID")),
+ ClientSecret: strings.TrimSpace(os.Getenv("YOUTUBE_OAUTH_CLIENT_SECRET")),
+ }
+ for _, value := range []string{config.ClientID, config.ClientSecret} {
+ if utf8.RuneCountInString(value) > maxYouTubeOAuthField {
+ return YouTubeOAuthConfig{}, errors.New("YouTube OAuth configuration value is too long")
+ }
+ }
+ if config.ClientID == "" && config.ClientSecret == "" {
+ return YouTubeOAuthConfig{}, nil
+ }
+ if config.ClientID == "" || config.ClientSecret == "" {
+ return YouTubeOAuthConfig{}, errors.New("YOUTUBE_OAUTH_CLIENT_ID and YOUTUBE_OAUTH_CLIENT_SECRET must be configured together")
+ }
+ return config, nil
+}
+
+func (c YouTubeOAuthConfig) Enabled() bool { return c.ClientID != "" }
+
+// YouTubeTokenResponse는 Google 토큰 엔드포인트 응답이다(2026-08-09 실측 기준).
+type YouTubeTokenResponse struct {
+ AccessToken string `json:"access_token"`
+ RefreshToken string `json:"refresh_token"`
+ ExpiresIn int64 `json:"expires_in"`
+ // RefreshTokenExpiresIn은 앱 게시 상태가 Testing일 때만 온다(실측 604799
+ // = 7일). 0이면 refresh token은 무기한이다.
+ RefreshTokenExpiresIn int64 `json:"refresh_token_expires_in"`
+ Scope string `json:"scope"`
+ TokenType string `json:"token_type"`
+}
+
+type YouTubeChannel struct {
+ ID string `json:"id"`
+ Title string `json:"title"`
+}
+
+// YouTubeAuthorizer는 Google OAuth·YouTube Data API와의 통신 계약이다.
+// 서비스 계층 테스트에서 대역으로 대체된다.
+type YouTubeAuthorizer interface {
+ // Exchange의 redirectURI는 코드 출처별 요구값(CodeSource.exchangeRedirectURI)
+ // 이며, 빈 문자열이면 파라미터를 생략한다.
+ Exchange(ctx context.Context, code, redirectURI string) (YouTubeTokenResponse, error)
+ RefreshAccessToken(ctx context.Context, refreshToken string) (YouTubeTokenResponse, error)
+ ChannelForToken(ctx context.Context, accessToken string) (YouTubeChannel, error)
+ // WebClientID는 웹 클라이언트(GIS)가 쓸 공개 클라이언트 식별자다.
+ WebClientID() string
+}
+
+type youtubeOAuthClient struct {
+ config YouTubeOAuthConfig
+ httpClient *http.Client
+ tokenURL string
+ channelsURL string
+}
+
+func NewYouTubeOAuthClient(config YouTubeOAuthConfig) (*youtubeOAuthClient, error) {
+ if !config.Enabled() {
+ return nil, errors.New("YouTube OAuth is not configured")
+ }
+ return &youtubeOAuthClient{
+ config: config,
+ httpClient: &http.Client{Timeout: 10 * time.Second},
+ tokenURL: googleOAuthTokenEndpoint,
+ channelsURL: youtubeChannelsEndpoint,
+ }, nil
+}
+
+// Exchange는 클라이언트가 획득한 인가 코드를 토큰으로 교환한다. 코드에 '/'
+// 등 예약 문자가 들어오므로(실측) 반드시 form 인코딩을 거친다.
+func (c *youtubeOAuthClient) Exchange(ctx context.Context, code, redirectURI string) (YouTubeTokenResponse, error) {
+ form := url.Values{
+ "client_id": {c.config.ClientID},
+ "client_secret": {c.config.ClientSecret},
+ "code": {strings.TrimSpace(code)},
+ "grant_type": {"authorization_code"},
+ }
+ if redirectURI != "" {
+ form.Set("redirect_uri", redirectURI)
+ }
+ return c.requestToken(ctx, form)
+}
+
+func (c *youtubeOAuthClient) WebClientID() string { return c.config.ClientID }
+
+// RefreshAccessToken은 refresh token으로 새 access token을 받는다. Google은
+// refresh token을 회전하지 않는 것이 기본이지만, 응답에 새 값이 오면 호출자가
+// 교체 저장해야 한다.
+func (c *youtubeOAuthClient) RefreshAccessToken(ctx context.Context, refreshToken string) (YouTubeTokenResponse, error) {
+ form := url.Values{
+ "client_id": {c.config.ClientID},
+ "client_secret": {c.config.ClientSecret},
+ "refresh_token": {strings.TrimSpace(refreshToken)},
+ "grant_type": {"refresh_token"},
+ }
+ return c.requestToken(ctx, form)
+}
+
+func (c *youtubeOAuthClient) requestToken(ctx context.Context, form url.Values) (YouTubeTokenResponse, error) {
+ request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.tokenURL, strings.NewReader(form.Encode()))
+ if err != nil {
+ return YouTubeTokenResponse{}, err
+ }
+ request.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+ response, err := c.httpClient.Do(request)
+ if err != nil {
+ return YouTubeTokenResponse{}, fmt.Errorf("request Google token endpoint: %w", err)
+ }
+ defer response.Body.Close()
+ if response.StatusCode != http.StatusOK {
+ _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 8<<10))
+ // 4xx는 코드 자체의 문제(만료·재사용·위조)라 클라이언트 오류로,
+ // 그 외는 Google 쪽 장애로 구분한다.
+ if response.StatusCode >= 400 && response.StatusCode < 500 {
+ return YouTubeTokenResponse{}, fmt.Errorf("%w: HTTP %d", ErrYouTubeAuthCodeRejected, response.StatusCode)
+ }
+ return YouTubeTokenResponse{}, fmt.Errorf("%w: HTTP %d", ErrYouTubeTokenExchange, response.StatusCode)
+ }
+ var result YouTubeTokenResponse
+ if err := json.NewDecoder(io.LimitReader(response.Body, 32<<10)).Decode(&result); err != nil {
+ return YouTubeTokenResponse{}, fmt.Errorf("decode Google token response: %w", err)
+ }
+ if strings.TrimSpace(result.AccessToken) == "" {
+ return YouTubeTokenResponse{}, fmt.Errorf("%w: response has no access_token", ErrYouTubeTokenExchange)
+ }
+ return result, nil
+}
+
+// ChannelForToken은 토큰이 매핑되는 YouTube 채널을 식별한다. 채널이 없는
+// Google 계정(실사용 가능 시나리오)은 ErrYouTubeChannelMissing으로 구분한다.
+func (c *youtubeOAuthClient) ChannelForToken(ctx context.Context, accessToken string) (YouTubeChannel, error) {
+ request, err := http.NewRequestWithContext(ctx, http.MethodGet, c.channelsURL+"?part=snippet&mine=true", nil)
+ if err != nil {
+ return YouTubeChannel{}, err
+ }
+ request.Header.Set("Authorization", "Bearer "+accessToken)
+ response, err := c.httpClient.Do(request)
+ if err != nil {
+ return YouTubeChannel{}, fmt.Errorf("request YouTube channels: %w", err)
+ }
+ defer response.Body.Close()
+ if response.StatusCode != http.StatusOK {
+ _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 8<<10))
+ return YouTubeChannel{}, fmt.Errorf("YouTube channels.list returned HTTP %d", response.StatusCode)
+ }
+ 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 YouTubeChannel{}, fmt.Errorf("decode YouTube channels response: %w", err)
+ }
+ if len(payload.Items) == 0 {
+ return YouTubeChannel{}, ErrYouTubeChannelMissing
+ }
+ item := payload.Items[0]
+ if strings.TrimSpace(item.ID) == "" {
+ return YouTubeChannel{}, errors.New("YouTube channels response has no channel id")
+ }
+ return YouTubeChannel{ID: item.ID, Title: item.Snippet.Title}, nil
+}
+
+// YouTubeConnectService는 클라이언트(네이티브 SDK 또는 웹 GIS 팝업)가 획득한
+// 인가 코드로 계정 연결을 완결한다. 브라우저 리다이렉트가 없으므로 state가
+// 필요 없고, 사용자 바인딩은 connect 엔드포인트의 Bearer 인증이 담당한다.
+type YouTubeConnectService struct {
+ oauth YouTubeAuthorizer
+ store StreamingAccountStore
+ users UserStatusChecker
+ cipher *ProviderTokenCipher
+ now func() time.Time
+}
+
+func NewYouTubeConnectService(oauth YouTubeAuthorizer, store StreamingAccountStore, users UserStatusChecker, cipher *ProviderTokenCipher) (*YouTubeConnectService, error) {
+ if oauth == nil || store == nil || users == nil || cipher == nil {
+ return nil, errors.New("YouTube connect dependencies must not be nil")
+ }
+ return &YouTubeConnectService{
+ oauth: oauth,
+ store: store,
+ users: users,
+ cipher: cipher,
+ now: func() time.Time { return time.Now().UTC() },
+ }, nil
+}
+
+// ConnectWithAuthCode는 인가 코드를 토큰으로 교환하고 채널을 식별해
+// 연결을 저장한다. connect 엔드포인트의 인라인 Bearer 인증은 사용자 상태를
+// 확인하지 않으므로 여기서 active를 확인한다.
+func (s *YouTubeConnectService) ConnectWithAuthCode(ctx context.Context, userID uuid.UUID, code string, source CodeSource) (YouTubeChannel, error) {
+ if err := s.ensureActive(ctx, userID); err != nil {
+ return YouTubeChannel{}, err
+ }
+ token, err := s.oauth.Exchange(ctx, code, source.exchangeRedirectURI())
+ if err != nil {
+ return YouTubeChannel{}, err
+ }
+ // refresh token 없이 연결을 저장하면 access token 만료(1시간) 후 송출이
+ // 조용히 죽는다. prompt=consent로 항상 와야 하며, 안 왔다면 연결 실패다.
+ if strings.TrimSpace(token.RefreshToken) == "" {
+ return YouTubeChannel{}, fmt.Errorf("%w: response has no refresh_token", ErrYouTubeTokenExchange)
+ }
+ channel, err := s.oauth.ChannelForToken(ctx, token.AccessToken)
+ if err != nil {
+ return YouTubeChannel{}, err
+ }
+ ciphertext, version, err := s.cipher.Encrypt(token.RefreshToken)
+ if err != nil {
+ return YouTubeChannel{}, err
+ }
+ account := StreamingAccount{
+ UserID: userID,
+ Provider: StreamingProviderYouTube,
+ ChannelID: channel.ID,
+ ChannelTitle: googleOptionalString(channel.Title, 255),
+ RefreshTokenCiphertext: ciphertext,
+ TokenKeyVersion: version,
+ RefreshTokenExpiresAt: s.refreshTokenExpiry(token),
+ }
+ if err := s.store.Upsert(ctx, account); err != nil {
+ return YouTubeChannel{}, fmt.Errorf("persist streaming account: %w", err)
+ }
+ return channel, nil
+}
+
+// WebClientID는 웹(GIS) 클라이언트가 initCodeClient에 쓸 공개 클라이언트
+// 식별자다 — 비밀이 아니며 GET /auth/youtube/config 로 노출된다.
+func (s *YouTubeConnectService) WebClientID() string {
+ return s.oauth.WebClientID()
+}
+
+func (s *YouTubeConnectService) 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
+}
+
+func (s *YouTubeConnectService) refreshTokenExpiry(token YouTubeTokenResponse) *time.Time {
+ if token.RefreshTokenExpiresIn <= 0 {
+ return nil
+ }
+ expiresAt := s.now().Add(time.Duration(token.RefreshTokenExpiresIn) * time.Second)
+ return &expiresAt
+}
+
+// YouTubeAccessTokenProvider는 저장된 refresh token으로 access token을
+// 발급·캐시한다. 갱신은 사용자 단위로 직렬화한다 — 한 사용자의 방송 세션
+// 여러 개가 동시에 만료를 만나도 토큰 엔드포인트 호출은 한 번만 나간다.
+type YouTubeAccessTokenProvider struct {
+ oauth YouTubeAuthorizer
+ store StreamingAccountStore
+ cipher *ProviderTokenCipher
+ now func() time.Time
+
+ mu sync.Mutex
+ users map[uuid.UUID]*userAccessToken
+}
+
+type userAccessToken struct {
+ mu sync.Mutex
+ token string
+ expiresAt time.Time
+}
+
+func NewYouTubeAccessTokenProvider(oauth YouTubeAuthorizer, store StreamingAccountStore, cipher *ProviderTokenCipher) (*YouTubeAccessTokenProvider, error) {
+ if oauth == nil || store == nil || cipher == nil {
+ return nil, errors.New("YouTube token provider dependencies must not be nil")
+ }
+ return &YouTubeAccessTokenProvider{
+ oauth: oauth,
+ store: store,
+ cipher: cipher,
+ now: func() time.Time { return time.Now().UTC() },
+ users: make(map[uuid.UUID]*userAccessToken),
+ }, nil
+}
+
+// AccessToken은 유효한 access token을 돌려준다. 캐시가 만료 여유(60초) 안에
+// 있으면 그대로 쓰고, 아니면 refresh token으로 갱신한다.
+func (p *YouTubeAccessTokenProvider) AccessToken(ctx context.Context, userID uuid.UUID) (string, error) {
+ state := p.userState(userID)
+ // 사용자 단위 락: 같은 사용자의 동시 호출은 첫 갱신을 기다렸다가 캐시를
+ // 재사용한다. 다른 사용자끼리는 서로 막지 않는다.
+ state.mu.Lock()
+ defer state.mu.Unlock()
+ if state.token != "" && p.now().Before(state.expiresAt.Add(-accessTokenExpirySlack)) {
+ return state.token, nil
+ }
+ account, err := p.store.Get(ctx, userID, StreamingProviderYouTube)
+ if err != nil {
+ if errors.Is(err, ErrStreamingAccountNotFound) {
+ return "", ErrStreamingNotConnected
+ }
+ return "", err
+ }
+ if len(account.RefreshTokenCiphertext) == 0 {
+ return "", ErrStreamingNotConnected
+ }
+ refreshToken, err := p.cipher.Decrypt(account.RefreshTokenCiphertext, account.TokenKeyVersion)
+ if err != nil {
+ return "", err
+ }
+ response, err := p.oauth.RefreshAccessToken(ctx, refreshToken)
+ if err != nil {
+ return "", err
+ }
+ // Google이 예외적으로 새 refresh token을 주면 행 락 하에 교체 저장한다.
+ if next := strings.TrimSpace(response.RefreshToken); next != "" && next != refreshToken {
+ ciphertext, version, err := p.cipher.Encrypt(next)
+ if err != nil {
+ return "", err
+ }
+ var expiresAt *time.Time
+ if response.RefreshTokenExpiresIn > 0 {
+ at := p.now().Add(time.Duration(response.RefreshTokenExpiresIn) * time.Second)
+ expiresAt = &at
+ }
+ if err := p.store.UpdateRefreshToken(ctx, account.ID, ciphertext, version, expiresAt); err != nil {
+ return "", fmt.Errorf("persist rotated refresh token: %w", err)
+ }
+ }
+ state.token = response.AccessToken
+ state.expiresAt = p.now().Add(time.Duration(response.ExpiresIn) * time.Second)
+ return state.token, nil
+}
+
+func (p *YouTubeAccessTokenProvider) userState(userID uuid.UUID) *userAccessToken {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ state := p.users[userID]
+ if state == nil {
+ state = &userAccessToken{}
+ p.users[userID] = state
+ }
+ return state
+}
diff --git a/internal/auth/youtube_oauth_http.go b/internal/auth/youtube_oauth_http.go
new file mode 100644
index 0000000..e7a7aa3
--- /dev/null
+++ b/internal/auth/youtube_oauth_http.go
@@ -0,0 +1,108 @@
+package auth
+
+import (
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "strings"
+
+ "github.com/google/uuid"
+)
+
+const maxYouTubeConnectRequestBody = 16 << 10
+
+// handleYouTubeConnect는 네이티브 클라이언트(GoogleSignIn SDK)가 획득한
+// serverAuthCode를 받아 YouTube 계정 연결을 완결한다. 브라우저 리다이렉트
+// 없이 XHR 1회로 끝나므로 콜백·state가 없고, 사용자 바인딩은 Bearer 인증이
+// 담당한다.
+func (h *tokenHTTPHandler) handleYouTubeConnect(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodOptions {
+ w.WriteHeader(http.StatusNoContent)
+ return
+ }
+ raw, ok := accessBearerToken(r)
+ if !ok {
+ h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.")
+ return
+ }
+ claims, err := h.service.ValidateAccessToken(raw)
+ if err != nil {
+ h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.")
+ return
+ }
+ userID, err := uuid.Parse(claims.Subject)
+ if err != nil {
+ h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.")
+ return
+ }
+ code, source, err := decodeYouTubeConnectRequest(w, r)
+ if err != nil {
+ h.writeError(w, r, http.StatusBadRequest, "bad_request", "Invalid YouTube connect request.")
+ return
+ }
+ channel, err := h.youtube.ConnectWithAuthCode(r.Context(), userID, code, source)
+ if err != nil {
+ switch {
+ case errors.Is(err, ErrUserInactive):
+ h.writeError(w, r, http.StatusUnauthorized, "unauthorized", "Authentication is required.")
+ case errors.Is(err, ErrYouTubeAuthCodeRejected):
+ h.writeError(w, r, http.StatusBadRequest, "invalid_auth_code", "The authorization code was rejected. Sign in with Google again.")
+ case errors.Is(err, ErrYouTubeChannelMissing):
+ h.writeError(w, r, http.StatusUnprocessableEntity, "youtube_channel_missing", "The Google account has no YouTube channel.")
+ case errors.Is(err, ErrYouTubeTokenExchange):
+ h.writeError(w, r, http.StatusBadGateway, "youtube_token_exchange_failed", "YouTube authorization could not be completed.")
+ default:
+ h.logger.Error("YouTube connect 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, map[string]any{
+ "connected": true,
+ "provider": StreamingProviderYouTube,
+ "channel": channel,
+ })
+}
+
+func decodeYouTubeConnectRequest(w http.ResponseWriter, r *http.Request) (string, CodeSource, error) {
+ request := struct {
+ ServerAuthCode string `json:"server_auth_code"`
+ CodeSource string `json:"code_source"`
+ }{}
+ r.Body = http.MaxBytesReader(w, r.Body, maxYouTubeConnectRequestBody)
+ decoder := json.NewDecoder(r.Body)
+ decoder.DisallowUnknownFields()
+ if err := decoder.Decode(&request); err != nil {
+ return "", "", err
+ }
+ if err := decoder.Decode(&struct{}{}); err != io.EOF {
+ return "", "", errors.New("request body must contain one JSON object")
+ }
+ request.ServerAuthCode = strings.TrimSpace(request.ServerAuthCode)
+ if request.ServerAuthCode == "" {
+ return "", "", errors.New("server_auth_code is required")
+ }
+ // code_source는 선택 필드다 — 생략 시 native(기존 계약 하위호환).
+ source := CodeSource(strings.TrimSpace(request.CodeSource))
+ if source == "" {
+ source = CodeSourceNative
+ }
+ if !source.Valid() {
+ return "", "", errors.New("code_source must be native or web_popup")
+ }
+ return request.ServerAuthCode, source, nil
+}
+
+// handleYouTubeConfig는 웹 클라이언트가 GIS 팝업을 초기화하는 데 필요한
+// 공개 설정을 준다. client_id는 공개 식별자라 인증이 필요 없다.
+func (h *tokenHTTPHandler) handleYouTubeConfig(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodOptions {
+ w.WriteHeader(http.StatusNoContent)
+ return
+ }
+ h.writeJSON(w, http.StatusOK, map[string]string{
+ "web_client_id": h.youtube.WebClientID(),
+ "scope": YouTubeStreamingScope,
+ })
+}
diff --git a/internal/auth/youtube_oauth_http_test.go b/internal/auth/youtube_oauth_http_test.go
new file mode 100644
index 0000000..7557f39
--- /dev/null
+++ b/internal/auth/youtube_oauth_http_test.go
@@ -0,0 +1,266 @@
+package auth
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "log/slog"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/google/uuid"
+)
+
+func testYouTubeHTTPHandler(t *testing.T, service *YouTubeConnectService) (*TokenService, http.Handler) {
+ t.Helper()
+ config, err := NewTokenHTTPConfig(false, []string{"http://localhost:3000"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ tokens := testTokenService(newMemoryRefreshStore())
+ handler := MountAuthHTTPWithServices(http.NotFoundHandler(), tokens, nil, nil, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil)), config, service)
+ return tokens, handler
+}
+
+func youtubeConnectRequest(t *testing.T, accessToken, body string) *http.Request {
+ t.Helper()
+ request := httptest.NewRequest(http.MethodPost, "/auth/youtube/connect", bytes.NewBufferString(body))
+ request.Header.Set("Content-Type", "application/json")
+ if accessToken != "" {
+ request.Header.Set("Authorization", "Bearer "+accessToken)
+ }
+ return request
+}
+
+func TestYouTubeConnectHTTPRequiresBearer(t *testing.T) {
+ service := testYouTubeConnectService(t, &stubYouTubeAuthorizer{}, newMemoryStreamingAccountStore(), UserStatusActive)
+ _, handler := testYouTubeHTTPHandler(t, service)
+
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, youtubeConnectRequest(t, "", `{"server_auth_code":"code"}`))
+ if response.Code != http.StatusUnauthorized {
+ t.Fatalf("status = %d, body = %s", response.Code, response.Body.String())
+ }
+}
+
+func TestYouTubeConnectHTTPCompletesConnection(t *testing.T) {
+ oauth := &stubYouTubeAuthorizer{
+ token: YouTubeTokenResponse{AccessToken: "at", RefreshToken: "rt", ExpiresIn: 3599},
+ channel: YouTubeChannel{ID: "UCabc", Title: "Team Framework"},
+ }
+ store := newMemoryStreamingAccountStore()
+ service := testYouTubeConnectService(t, oauth, store, UserStatusActive)
+ tokens, handler := testYouTubeHTTPHandler(t, service)
+
+ userID := uuid.New()
+ pair, err := tokens.IssuePair(context.Background(), userID, ClientInfo{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, youtubeConnectRequest(t, pair.AccessToken, `{"server_auth_code":"4/0AXserver-code"}`))
+ if response.Code != http.StatusOK {
+ t.Fatalf("status = %d, body = %s", response.Code, response.Body.String())
+ }
+ var payload struct {
+ Connected bool `json:"connected"`
+ Provider string `json:"provider"`
+ Channel struct {
+ ID string `json:"id"`
+ Title string `json:"title"`
+ } `json:"channel"`
+ }
+ if err := json.NewDecoder(response.Body).Decode(&payload); err != nil {
+ t.Fatal(err)
+ }
+ if !payload.Connected || payload.Provider != "youtube" || payload.Channel.ID != "UCabc" {
+ t.Fatalf("payload = %+v", payload)
+ }
+ if _, err := store.Get(context.Background(), userID, StreamingProviderYouTube); err != nil {
+ t.Fatalf("connection not persisted: %v", err)
+ }
+}
+
+func TestYouTubeConnectHTTPBadRequests(t *testing.T) {
+ service := testYouTubeConnectService(t, &stubYouTubeAuthorizer{}, newMemoryStreamingAccountStore(), UserStatusActive)
+ tokens, handler := testYouTubeHTTPHandler(t, service)
+ pair, err := tokens.IssuePair(context.Background(), uuid.New(), ClientInfo{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ for name, body := range map[string]string{
+ "empty body": ``,
+ "missing code": `{}`,
+ "blank code": `{"server_auth_code":" "}`,
+ "unknown field": `{"server_auth_code":"x","extra":true}`,
+ } {
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, youtubeConnectRequest(t, pair.AccessToken, body))
+ if response.Code != http.StatusBadRequest {
+ t.Fatalf("%s: status = %d, want 400 (body: %s)", name, response.Code, response.Body.String())
+ }
+ }
+}
+
+func TestYouTubeConnectHTTPMapsExchangeErrors(t *testing.T) {
+ cases := []struct {
+ name string
+ stub *stubYouTubeAuthorizer
+ wantStatus int
+ wantCode string
+ }{
+ {
+ name: "rejected code",
+ stub: &stubYouTubeAuthorizer{exchangeErr: ErrYouTubeAuthCodeRejected},
+ wantStatus: http.StatusBadRequest,
+ wantCode: "invalid_auth_code",
+ },
+ {
+ name: "google outage",
+ stub: &stubYouTubeAuthorizer{exchangeErr: ErrYouTubeTokenExchange},
+ wantStatus: http.StatusBadGateway,
+ wantCode: "youtube_token_exchange_failed",
+ },
+ {
+ name: "channel missing",
+ stub: &stubYouTubeAuthorizer{
+ token: YouTubeTokenResponse{AccessToken: "at", RefreshToken: "rt"},
+ channelErr: ErrYouTubeChannelMissing,
+ },
+ wantStatus: http.StatusUnprocessableEntity,
+ wantCode: "youtube_channel_missing",
+ },
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ service := testYouTubeConnectService(t, tc.stub, newMemoryStreamingAccountStore(), UserStatusActive)
+ tokens, handler := testYouTubeHTTPHandler(t, service)
+ pair, err := tokens.IssuePair(context.Background(), uuid.New(), ClientInfo{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, youtubeConnectRequest(t, pair.AccessToken, `{"server_auth_code":"code"}`))
+ if response.Code != tc.wantStatus {
+ t.Fatalf("status = %d, want %d (body: %s)", response.Code, tc.wantStatus, response.Body.String())
+ }
+ var payload struct {
+ Error struct {
+ Code string `json:"code"`
+ } `json:"error"`
+ }
+ if err := json.NewDecoder(response.Body).Decode(&payload); err != nil {
+ t.Fatal(err)
+ }
+ if payload.Error.Code != tc.wantCode {
+ t.Fatalf("error code = %q, want %q", payload.Error.Code, tc.wantCode)
+ }
+ })
+ }
+}
+
+// TestYouTubeConnectHTTPCodeSource: code_source가 교환 redirect_uri 매핑으로
+// 이어지고(web_popup→postmessage 실측 계약), 미지 값은 400이어야 한다.
+func TestYouTubeConnectHTTPCodeSource(t *testing.T) {
+ oauth := &stubYouTubeAuthorizer{
+ token: YouTubeTokenResponse{AccessToken: "at", RefreshToken: "rt", ExpiresIn: 3599},
+ channel: YouTubeChannel{ID: "UCabc"},
+ }
+ service := testYouTubeConnectService(t, oauth, newMemoryStreamingAccountStore(), UserStatusActive)
+ tokens, handler := testYouTubeHTTPHandler(t, service)
+ pair, err := tokens.IssuePair(context.Background(), uuid.New(), ClientInfo{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, youtubeConnectRequest(t, pair.AccessToken, `{"server_auth_code":"c","code_source":"web_popup"}`))
+ if response.Code != http.StatusOK {
+ t.Fatalf("web_popup status = %d, body = %s", response.Code, response.Body.String())
+ }
+ oauth.mu.Lock()
+ redirect := oauth.exchangeRedirect
+ oauth.mu.Unlock()
+ if redirect != "postmessage" {
+ t.Fatalf("web_popup exchanged with redirect %q, want postmessage", redirect)
+ }
+
+ // 생략 시 native(하위호환) — redirect_uri 생략.
+ response = httptest.NewRecorder()
+ handler.ServeHTTP(response, youtubeConnectRequest(t, pair.AccessToken, `{"server_auth_code":"c2"}`))
+ if response.Code != http.StatusOK {
+ t.Fatalf("default source status = %d, body = %s", response.Code, response.Body.String())
+ }
+ oauth.mu.Lock()
+ redirect = oauth.exchangeRedirect
+ oauth.mu.Unlock()
+ if redirect != "" {
+ t.Fatalf("native exchanged with redirect %q, want omitted", redirect)
+ }
+
+ // 미지 출처는 400.
+ response = httptest.NewRecorder()
+ handler.ServeHTTP(response, youtubeConnectRequest(t, pair.AccessToken, `{"server_auth_code":"c3","code_source":"browser"}`))
+ if response.Code != http.StatusBadRequest {
+ t.Fatalf("unknown source status = %d, want 400", response.Code)
+ }
+}
+
+// TestYouTubeConfigHTTPExposesWebClientID: 웹 클라이언트가 GIS 초기화에 쓸
+// 공개 설정 — 인증 없이 접근 가능해야 한다.
+func TestYouTubeConfigHTTPExposesWebClientID(t *testing.T) {
+ service := testYouTubeConnectService(t, &stubYouTubeAuthorizer{}, newMemoryStreamingAccountStore(), UserStatusActive)
+ _, handler := testYouTubeHTTPHandler(t, service)
+
+ request := httptest.NewRequest(http.MethodGet, "/auth/youtube/config", nil)
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, request)
+ if response.Code != http.StatusOK {
+ t.Fatalf("status = %d, body = %s", response.Code, response.Body.String())
+ }
+ var payload struct {
+ WebClientID string `json:"web_client_id"`
+ Scope string `json:"scope"`
+ }
+ if err := json.NewDecoder(response.Body).Decode(&payload); err != nil {
+ t.Fatal(err)
+ }
+ if payload.WebClientID != "stub-web-client-id" || payload.Scope != YouTubeStreamingScope {
+ t.Fatalf("payload = %+v", payload)
+ }
+}
+
+// TestYouTubeCallbackRouteRemoved: serverAuthCode 전환으로 브라우저 콜백
+// 라우트는 존재하지 않아야 한다(다음 핸들러로 흘러 404).
+func TestYouTubeCallbackRouteRemoved(t *testing.T) {
+ service := testYouTubeConnectService(t, &stubYouTubeAuthorizer{}, newMemoryStreamingAccountStore(), UserStatusActive)
+ _, handler := testYouTubeHTTPHandler(t, service)
+ request := httptest.NewRequest(http.MethodGet, "/auth/youtube/callback?state=x&code=y", nil)
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, request)
+ if response.Code != http.StatusNotFound {
+ t.Fatalf("status = %d, want 404 (callback route must not exist)", response.Code)
+ }
+}
+
+// TestYouTubeRoutesAbsentWhenServiceNil: YouTube 서비스가 조립되지 않은
+// 배포에서는 라우트 자체가 등록되지 않아야 한다(기존 mount 계약 보존).
+func TestYouTubeRoutesAbsentWhenServiceNil(t *testing.T) {
+ config, err := NewTokenHTTPConfig(false, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tokens := testTokenService(newMemoryRefreshStore())
+ handler := MountAuthHTTPWithServices(http.NotFoundHandler(), tokens, nil, nil, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil)), config)
+
+ request := httptest.NewRequest(http.MethodPost, "/auth/youtube/connect", nil)
+ response := httptest.NewRecorder()
+ handler.ServeHTTP(response, request)
+ if response.Code != http.StatusNotFound {
+ t.Fatalf("status = %d, want 404 when service is absent", response.Code)
+ }
+}
diff --git a/internal/auth/youtube_oauth_test.go b/internal/auth/youtube_oauth_test.go
new file mode 100644
index 0000000..a065ed2
--- /dev/null
+++ b/internal/auth/youtube_oauth_test.go
@@ -0,0 +1,510 @@
+package auth
+
+import (
+ "bytes"
+ "context"
+ "encoding/base64"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+)
+
+func testProviderTokenCipher(t *testing.T) *ProviderTokenCipher {
+ t.Helper()
+ cipher, err := NewProviderTokenCipherFromBase64(base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{7}, 32)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return cipher
+}
+
+func TestLoadYouTubeOAuthConfigFromEnv(t *testing.T) {
+ t.Run("all empty disables", func(t *testing.T) {
+ t.Setenv("YOUTUBE_OAUTH_CLIENT_ID", "")
+ t.Setenv("YOUTUBE_OAUTH_CLIENT_SECRET", "")
+ config, err := LoadYouTubeOAuthConfigFromEnv()
+ if err != nil || config.Enabled() {
+ t.Fatalf("config = %+v, err = %v, want disabled without error", config, err)
+ }
+ })
+ t.Run("partial credentials is an error", func(t *testing.T) {
+ t.Setenv("YOUTUBE_OAUTH_CLIENT_ID", "client-id")
+ t.Setenv("YOUTUBE_OAUTH_CLIENT_SECRET", "")
+ if _, err := LoadYouTubeOAuthConfigFromEnv(); err == nil {
+ t.Fatal("partial credentials must fail")
+ }
+ })
+ t.Run("complete credentials enables", func(t *testing.T) {
+ t.Setenv("YOUTUBE_OAUTH_CLIENT_ID", "client-id")
+ t.Setenv("YOUTUBE_OAUTH_CLIENT_SECRET", "client-secret")
+ config, err := LoadYouTubeOAuthConfigFromEnv()
+ if err != nil || !config.Enabled() {
+ t.Fatalf("config = %+v, err = %v, want enabled", config, err)
+ }
+ })
+}
+
+// TestCodeSourceExchangeRedirectURI: 출처별 교환 redirect_uri 매핑 —
+// 2026-08-10 실측(웹 팝업 코드는 생략 시 "Missing parameter: redirect_uri",
+// postmessage로 성공)을 계약으로 고정한다.
+func TestCodeSourceExchangeRedirectURI(t *testing.T) {
+ if got := CodeSourceNative.exchangeRedirectURI(); got != "" {
+ t.Fatalf("native redirect = %q, want omitted", got)
+ }
+ if got := CodeSourceWebPopup.exchangeRedirectURI(); got != "postmessage" {
+ t.Fatalf("web_popup redirect = %q, want postmessage", got)
+ }
+ if CodeSource("browser").Valid() {
+ t.Fatal("unknown code source must be invalid")
+ }
+ if !CodeSourceNative.Valid() || !CodeSourceWebPopup.Valid() {
+ t.Fatal("known code sources must be valid")
+ }
+}
+
+// TestYouTubeExchangeRedirectURIParameter: redirectURI 인자가 빈 값이면 폼에서
+// 생략되고, 값이 있으면 포함되는지 검증한다.
+func TestYouTubeExchangeRedirectURIParameter(t *testing.T) {
+ const rawCode = "4/0AXtestCODEwith/slash"
+ var received url.Values
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if err := r.ParseForm(); err != nil {
+ t.Errorf("parse form: %v", err)
+ }
+ received = r.PostForm
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"access_token":"at-value","refresh_token":"rt-value","expires_in":3599,"scope":"s","token_type":"Bearer"}`))
+ }))
+ defer server.Close()
+
+ client, err := NewYouTubeOAuthClient(YouTubeOAuthConfig{ClientID: "id", ClientSecret: "secret"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ client.tokenURL = server.URL
+
+ t.Run("omitted for native", func(t *testing.T) {
+ response, err := client.Exchange(context.Background(), rawCode, CodeSourceNative.exchangeRedirectURI())
+ if err != nil {
+ t.Fatal(err)
+ }
+ // 실측에서 코드에 '/'가 포함돼 왔다 — form 인코딩 왕복 검증.
+ if received.Get("code") != rawCode {
+ t.Fatalf("server received code %q, want %q", received.Get("code"), rawCode)
+ }
+ if _, present := received["redirect_uri"]; present {
+ t.Fatal("redirect_uri must be omitted for native codes")
+ }
+ if response.AccessToken != "at-value" || response.RefreshToken != "rt-value" {
+ t.Fatalf("response = %+v", response)
+ }
+ })
+ t.Run("postmessage for web popup", func(t *testing.T) {
+ if _, err := client.Exchange(context.Background(), rawCode, CodeSourceWebPopup.exchangeRedirectURI()); err != nil {
+ t.Fatal(err)
+ }
+ if received.Get("redirect_uri") != "postmessage" {
+ t.Fatalf("redirect_uri = %q, want postmessage", received.Get("redirect_uri"))
+ }
+ })
+ if client.WebClientID() != "id" {
+ t.Fatalf("WebClientID = %q", client.WebClientID())
+ }
+}
+
+// TestYouTubeExchangeRejectedCode: Google이 4xx로 거절한 코드는 클라이언트
+// 오류(ErrYouTubeAuthCodeRejected)로, 5xx는 게이트웨이 오류로 구분돼야 한다.
+func TestYouTubeExchangeRejectedCode(t *testing.T) {
+ rejected := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusBadRequest)
+ _, _ = w.Write([]byte(`{"error":"invalid_grant"}`))
+ }))
+ defer rejected.Close()
+ broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(http.StatusBadGateway)
+ }))
+ defer broken.Close()
+
+ client, err := NewYouTubeOAuthClient(YouTubeOAuthConfig{ClientID: "id", ClientSecret: "secret"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ client.tokenURL = rejected.URL
+ if _, err := client.Exchange(context.Background(), "stale-code", ""); !errors.Is(err, ErrYouTubeAuthCodeRejected) {
+ t.Fatalf("4xx error = %v, want ErrYouTubeAuthCodeRejected", err)
+ }
+ client.tokenURL = broken.URL
+ if _, err := client.Exchange(context.Background(), "any-code", ""); !errors.Is(err, ErrYouTubeTokenExchange) {
+ t.Fatalf("5xx error = %v, want ErrYouTubeTokenExchange", err)
+ }
+}
+
+func TestYouTubeTokenResponseParsesTestingExpiry(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"access_token":"at","refresh_token":"rt","expires_in":3599,"refresh_token_expires_in":604799,"scope":"s","token_type":"Bearer"}`))
+ }))
+ defer server.Close()
+ client, err := NewYouTubeOAuthClient(YouTubeOAuthConfig{ClientID: "id", ClientSecret: "secret"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ client.tokenURL = server.URL
+ response, err := client.Exchange(context.Background(), "code", "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Testing 게시 상태에서만 오는 필드(실측 604799=7일). Production 발급분은
+ // 필드 자체가 없다(실측 2026-08-10) — 0으로 파싱되어 "무기한"으로 다룬다.
+ if response.RefreshTokenExpiresIn != 604799 {
+ t.Fatalf("refresh_token_expires_in = %d, want 604799", response.RefreshTokenExpiresIn)
+ }
+}
+
+func TestYouTubeChannelForToken(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if got := r.Header.Get("Authorization"); got != "Bearer at-value" {
+ t.Errorf("authorization header = %q", got)
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"items":[{"id":"UCabc","snippet":{"title":"Team Framework"}}]}`))
+ }))
+ defer server.Close()
+ empty := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"items":[]}`))
+ }))
+ defer empty.Close()
+
+ client, err := NewYouTubeOAuthClient(YouTubeOAuthConfig{ClientID: "id", ClientSecret: "secret"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ client.channelsURL = server.URL
+ channel, err := client.ChannelForToken(context.Background(), "at-value")
+ if err != nil || channel.ID != "UCabc" || channel.Title != "Team Framework" {
+ t.Fatalf("channel = %+v, err = %v", channel, err)
+ }
+
+ // 채널 없는 Google 계정(실사용 시나리오)은 전용 에러로 구분돼야 한다.
+ client.channelsURL = empty.URL
+ if _, err := client.ChannelForToken(context.Background(), "at-value"); !errors.Is(err, ErrYouTubeChannelMissing) {
+ t.Fatalf("empty items error = %v, want ErrYouTubeChannelMissing", err)
+ }
+}
+
+// ---- 서비스 계층 테스트 대역 ----
+
+type stubYouTubeAuthorizer struct {
+ mu sync.Mutex
+ exchangeCode string
+ exchangeRedirect string
+ token YouTubeTokenResponse
+ exchangeErr error
+ channel YouTubeChannel
+ channelErr error
+ refreshToken YouTubeTokenResponse
+ refreshErr error
+ refreshCalls atomic.Int64
+ refreshedWith string
+}
+
+func (s *stubYouTubeAuthorizer) Exchange(_ context.Context, code, redirectURI string) (YouTubeTokenResponse, error) {
+ s.mu.Lock()
+ s.exchangeCode = code
+ s.exchangeRedirect = redirectURI
+ s.mu.Unlock()
+ return s.token, s.exchangeErr
+}
+
+func (s *stubYouTubeAuthorizer) WebClientID() string { return "stub-web-client-id" }
+
+func (s *stubYouTubeAuthorizer) RefreshAccessToken(_ context.Context, refreshToken string) (YouTubeTokenResponse, error) {
+ s.refreshCalls.Add(1)
+ s.mu.Lock()
+ s.refreshedWith = refreshToken
+ s.mu.Unlock()
+ return s.refreshToken, s.refreshErr
+}
+
+func (s *stubYouTubeAuthorizer) ChannelForToken(context.Context, string) (YouTubeChannel, error) {
+ return s.channel, s.channelErr
+}
+
+type memoryStreamingAccountStore struct {
+ mu sync.Mutex
+ accounts map[string]StreamingAccount
+}
+
+func newMemoryStreamingAccountStore() *memoryStreamingAccountStore {
+ return &memoryStreamingAccountStore{accounts: make(map[string]StreamingAccount)}
+}
+
+func streamingKey(userID uuid.UUID, provider StreamingProvider) string {
+ return userID.String() + "/" + string(provider)
+}
+
+func (s *memoryStreamingAccountStore) Upsert(_ context.Context, account StreamingAccount) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ key := streamingKey(account.UserID, account.Provider)
+ if existing, ok := s.accounts[key]; ok {
+ account.ID = existing.ID
+ } else if account.ID == uuid.Nil {
+ account.ID = uuid.New()
+ }
+ s.accounts[key] = account
+ return nil
+}
+
+func (s *memoryStreamingAccountStore) Get(_ context.Context, userID uuid.UUID, provider StreamingProvider) (StreamingAccount, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ account, ok := s.accounts[streamingKey(userID, provider)]
+ if !ok {
+ return StreamingAccount{}, ErrStreamingAccountNotFound
+ }
+ return account, nil
+}
+
+func (s *memoryStreamingAccountStore) UpdateRefreshToken(_ context.Context, id uuid.UUID, ciphertext []byte, version *int16, expiresAt *time.Time) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ for key, account := range s.accounts {
+ if account.ID == id {
+ account.RefreshTokenCiphertext = ciphertext
+ account.TokenKeyVersion = version
+ account.RefreshTokenExpiresAt = expiresAt
+ 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()
+ for key, account := range s.accounts {
+ if account.ID == id {
+ account.StreamID = &info.StreamID
+ account.IngestionAddress = &info.IngestionAddress
+ account.BackupIngestionAddress = &info.BackupIngestionAddress
+ account.RtmpsIngestionAddress = &info.RtmpsIngestionAddress
+ account.RtmpsBackupIngestionAddress = &info.RtmpsBackupIngestionAddress
+ account.StreamNameCiphertext = info.StreamNameCiphertext
+ account.StreamNameKeyVersion = info.StreamNameKeyVersion
+ s.accounts[key] = account
+ return nil
+ }
+ }
+ return ErrStreamingAccountNotFound
+}
+
+func testYouTubeConnectService(t *testing.T, oauth YouTubeAuthorizer, store StreamingAccountStore, status UserStatus) *YouTubeConnectService {
+ t.Helper()
+ service, err := NewYouTubeConnectService(oauth, store, testUserStatusChecker{status: status}, testProviderTokenCipher(t))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return service
+}
+
+func TestYouTubeConnectWithAuthCodePersistsEncryptedConnection(t *testing.T) {
+ oauth := &stubYouTubeAuthorizer{
+ token: YouTubeTokenResponse{
+ AccessToken: "at-value",
+ RefreshToken: "rt-secret",
+ ExpiresIn: 3599,
+ RefreshTokenExpiresIn: 604799,
+ },
+ channel: YouTubeChannel{ID: "UCabc", Title: "Team Framework"},
+ }
+ store := newMemoryStreamingAccountStore()
+ service := testYouTubeConnectService(t, oauth, store, UserStatusActive)
+ userID := uuid.New()
+
+ channel, err := service.ConnectWithAuthCode(context.Background(), userID, "server-auth-code", CodeSourceNative)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if channel.ID != "UCabc" {
+ t.Fatalf("channel = %+v", channel)
+ }
+ oauth.mu.Lock()
+ exchanged := oauth.exchangeCode
+ oauth.mu.Unlock()
+ if exchanged != "server-auth-code" {
+ t.Fatalf("exchanged code = %q", exchanged)
+ }
+
+ account, err := store.Get(context.Background(), userID, StreamingProviderYouTube)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if account.ChannelID != "UCabc" || account.ChannelTitle == nil || *account.ChannelTitle != "Team Framework" {
+ t.Fatalf("account channel = %q/%v", account.ChannelID, account.ChannelTitle)
+ }
+ // refresh token은 평문이 아니라 암호문으로 저장돼야 한다.
+ if bytes.Contains(account.RefreshTokenCiphertext, []byte("rt-secret")) {
+ t.Fatal("refresh token stored in plaintext")
+ }
+ plaintext, err := testProviderTokenCipher(t).Decrypt(account.RefreshTokenCiphertext, account.TokenKeyVersion)
+ if err != nil || plaintext != "rt-secret" {
+ t.Fatalf("decrypt = (%q, %v)", plaintext, err)
+ }
+ // Testing 게시 상태 토큰의 만료(7일)는 재연결 유도 신호로 추적돼야 한다.
+ if account.RefreshTokenExpiresAt == nil {
+ t.Fatal("RefreshTokenExpiresAt must be tracked when refresh_token_expires_in is present")
+ }
+}
+
+func TestYouTubeConnectWithAuthCodeProductionTokenHasNoExpiry(t *testing.T) {
+ // Production 게시 발급분은 refresh_token_expires_in 부재(실측 2026-08-10)
+ // — 만료 추적 컬럼은 NULL이어야 한다.
+ oauth := &stubYouTubeAuthorizer{
+ token: YouTubeTokenResponse{AccessToken: "at", RefreshToken: "rt", ExpiresIn: 3599},
+ channel: YouTubeChannel{ID: "UCabc"},
+ }
+ store := newMemoryStreamingAccountStore()
+ service := testYouTubeConnectService(t, oauth, store, UserStatusActive)
+ userID := uuid.New()
+ if _, err := service.ConnectWithAuthCode(context.Background(), userID, "code", CodeSourceNative); err != nil {
+ t.Fatal(err)
+ }
+ account, err := store.Get(context.Background(), userID, StreamingProviderYouTube)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if account.RefreshTokenExpiresAt != nil {
+ t.Fatalf("RefreshTokenExpiresAt = %v, want nil for production tokens", account.RefreshTokenExpiresAt)
+ }
+}
+
+func TestYouTubeConnectWithAuthCodeRejectsInactiveUser(t *testing.T) {
+ service := testYouTubeConnectService(t, &stubYouTubeAuthorizer{}, newMemoryStreamingAccountStore(), UserStatusDisabled)
+ if _, err := service.ConnectWithAuthCode(context.Background(), uuid.New(), "code", CodeSourceNative); !errors.Is(err, ErrUserInactive) {
+ t.Fatalf("error = %v, want ErrUserInactive", err)
+ }
+}
+
+func TestYouTubeConnectWithAuthCodeRequiresRefreshToken(t *testing.T) {
+ oauth := &stubYouTubeAuthorizer{token: YouTubeTokenResponse{AccessToken: "at-only"}}
+ service := testYouTubeConnectService(t, oauth, newMemoryStreamingAccountStore(), UserStatusActive)
+ if _, err := service.ConnectWithAuthCode(context.Background(), uuid.New(), "code", CodeSourceNative); !errors.Is(err, ErrYouTubeTokenExchange) {
+ t.Fatalf("error = %v, want ErrYouTubeTokenExchange (missing refresh_token)", err)
+ }
+}
+
+// TestYouTubeAccessTokenProviderSerializesRefresh: 같은 사용자의 동시 요청이
+// 토큰 갱신을 한 번만 트리거하고 모두 같은 토큰을 받아야 한다(다중 방송
+// 세션 시나리오).
+func TestYouTubeAccessTokenProviderSerializesRefresh(t *testing.T) {
+ 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)
+ }
+ oauth := &stubYouTubeAuthorizer{refreshToken: YouTubeTokenResponse{AccessToken: "fresh-at", ExpiresIn: 3600}}
+ provider, err := NewYouTubeAccessTokenProvider(oauth, store, cipher)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ const parallel = 8
+ tokens := make([]string, parallel)
+ var wait sync.WaitGroup
+ for i := 0; i < parallel; i++ {
+ wait.Add(1)
+ go func(index int) {
+ defer wait.Done()
+ token, err := provider.AccessToken(context.Background(), userID)
+ if err != nil {
+ t.Errorf("AccessToken: %v", err)
+ return
+ }
+ tokens[index] = token
+ }(i)
+ }
+ wait.Wait()
+
+ if calls := oauth.refreshCalls.Load(); calls != 1 {
+ t.Fatalf("refresh calls = %d, want 1 (per-user serialization)", calls)
+ }
+ for _, token := range tokens {
+ if token != "fresh-at" {
+ t.Fatalf("tokens = %v, want all %q", tokens, "fresh-at")
+ }
+ }
+ oauth.mu.Lock()
+ refreshedWith := oauth.refreshedWith
+ oauth.mu.Unlock()
+ if refreshedWith != "rt-secret" {
+ t.Fatalf("refreshed with %q, want decrypted refresh token", refreshedWith)
+ }
+}
+
+func TestYouTubeAccessTokenProviderNotConnected(t *testing.T) {
+ provider, err := NewYouTubeAccessTokenProvider(&stubYouTubeAuthorizer{}, newMemoryStreamingAccountStore(), testProviderTokenCipher(t))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := provider.AccessToken(context.Background(), uuid.New()); !errors.Is(err, ErrStreamingNotConnected) {
+ t.Fatalf("error = %v, want ErrStreamingNotConnected", err)
+ }
+}
+
+// TestYouTubeAccessTokenProviderPersistsRotatedToken: Google이 예외적으로 새
+// refresh token을 돌려주면 암호화해 교체 저장해야 한다.
+func TestYouTubeAccessTokenProviderPersistsRotatedToken(t *testing.T) {
+ cipher := testProviderTokenCipher(t)
+ ciphertext, version, err := cipher.Encrypt("rt-old")
+ 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{refreshToken: YouTubeTokenResponse{AccessToken: "fresh-at", RefreshToken: "rt-new", ExpiresIn: 3600}}
+ provider, err := NewYouTubeAccessTokenProvider(oauth, store, cipher)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := provider.AccessToken(context.Background(), userID); err != nil {
+ t.Fatal(err)
+ }
+ account, err := store.Get(context.Background(), userID, StreamingProviderYouTube)
+ if err != nil {
+ t.Fatal(err)
+ }
+ plaintext, err := cipher.Decrypt(account.RefreshTokenCiphertext, account.TokenKeyVersion)
+ if err != nil || plaintext != "rt-new" {
+ t.Fatalf("stored refresh token = (%q, %v), want rt-new", plaintext, err)
+ }
+}
diff --git a/internal/config/config.go b/internal/config/config.go
index af54122..d9561d2 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -69,7 +69,6 @@ type Config struct {
UDPMuxPort int
DisconnectedGracePeriod time.Duration
FrameQueueSize int
- YoutubeStreamKey string
EnableAudioEgress bool
EgressLatencyLog bool
EgressAudioOffset time.Duration
@@ -111,7 +110,6 @@ func Load() (Config, error) {
AITimeout: envDuration("AI_GRPC_TIMEOUT", 5*time.Second),
PrivacyFixedDelay: envDurationWithMillisecondsAlias("AI_PRIVACY_FIXED_DELAY", "AI_PRIVACY_FIXED_DELAY_MS", 20*time.Millisecond),
FrameQueueSize: envInt("AI_FRAME_QUEUE_SIZE", 2),
- YoutubeStreamKey: strings.TrimSpace(os.Getenv("YOUTUBE_STREAM_KEY")),
EnableAudioEgress: envBool("ENABLE_AUDIO_EGRESS", false),
EgressLatencyLog: envBool("EGRESS_LATENCY_LOG", false),
EgressAudioOffset: time.Duration(envInt("EGRESS_AUDIO_OFFSET_MS", 0)) * time.Millisecond,
diff --git a/internal/database/migration/sql/000003_create_streaming_accounts.down.sql b/internal/database/migration/sql/000003_create_streaming_accounts.down.sql
new file mode 100644
index 0000000..3c0778a
--- /dev/null
+++ b/internal/database/migration/sql/000003_create_streaming_accounts.down.sql
@@ -0,0 +1,5 @@
+BEGIN;
+
+DROP TABLE IF EXISTS streaming_accounts;
+
+COMMIT;
diff --git a/internal/database/migration/sql/000003_create_streaming_accounts.up.sql b/internal/database/migration/sql/000003_create_streaming_accounts.up.sql
new file mode 100644
index 0000000..fc72d4f
--- /dev/null
+++ b/internal/database/migration/sql/000003_create_streaming_accounts.up.sql
@@ -0,0 +1,33 @@
+BEGIN;
+
+CREATE TABLE streaming_accounts (
+ id UUID PRIMARY KEY,
+ user_id UUID NOT NULL,
+ provider VARCHAR(20) NOT NULL,
+ channel_id VARCHAR(255) NOT NULL,
+ channel_title VARCHAR(255),
+ refresh_token_ciphertext BYTEA,
+ token_key_version SMALLINT,
+ refresh_token_expires_at TIMESTAMPTZ,
+ manual_ingest_url TEXT,
+ connected_at TIMESTAMPTZ NOT NULL,
+ created_at TIMESTAMPTZ NOT NULL,
+ updated_at TIMESTAMPTZ NOT NULL,
+
+ CONSTRAINT fk_streaming_accounts_user
+ FOREIGN KEY (user_id)
+ REFERENCES users (id)
+ ON UPDATE CASCADE
+ ON DELETE CASCADE,
+
+ CONSTRAINT chk_streaming_provider
+ CHECK (provider IN ('youtube', 'chzzk')),
+
+ CONSTRAINT uidx_streaming_user_provider
+ UNIQUE (user_id, provider)
+);
+
+CREATE INDEX idx_streaming_accounts_user_id
+ ON streaming_accounts (user_id);
+
+COMMIT;
diff --git a/internal/database/migration/sql/000004_add_streaming_stream_info.down.sql b/internal/database/migration/sql/000004_add_streaming_stream_info.down.sql
new file mode 100644
index 0000000..0c848a9
--- /dev/null
+++ b/internal/database/migration/sql/000004_add_streaming_stream_info.down.sql
@@ -0,0 +1,12 @@
+BEGIN;
+
+ALTER TABLE streaming_accounts
+ DROP COLUMN IF EXISTS stream_id,
+ DROP COLUMN IF EXISTS ingestion_address,
+ DROP COLUMN IF EXISTS backup_ingestion_address,
+ DROP COLUMN IF EXISTS rtmps_ingestion_address,
+ DROP COLUMN IF EXISTS rtmps_backup_ingestion_address,
+ DROP COLUMN IF EXISTS stream_name_ciphertext,
+ DROP COLUMN IF EXISTS stream_name_key_version;
+
+COMMIT;
diff --git a/internal/database/migration/sql/000004_add_streaming_stream_info.up.sql b/internal/database/migration/sql/000004_add_streaming_stream_info.up.sql
new file mode 100644
index 0000000..a2d00e6
--- /dev/null
+++ b/internal/database/migration/sql/000004_add_streaming_stream_info.up.sql
@@ -0,0 +1,12 @@
+BEGIN;
+
+ALTER TABLE streaming_accounts
+ ADD COLUMN stream_id VARCHAR(255),
+ ADD COLUMN ingestion_address TEXT,
+ ADD COLUMN backup_ingestion_address TEXT,
+ ADD COLUMN rtmps_ingestion_address TEXT,
+ ADD COLUMN rtmps_backup_ingestion_address TEXT,
+ ADD COLUMN stream_name_ciphertext BYTEA,
+ ADD COLUMN stream_name_key_version SMALLINT;
+
+COMMIT;
diff --git a/internal/media/track.go b/internal/media/track.go
index f30b13e..05a7065 100644
--- a/internal/media/track.go
+++ b/internal/media/track.go
@@ -7,6 +7,7 @@ import (
"io"
"log/slog"
"sync"
+ "sync/atomic"
"time"
"inno-live-server/internal/config"
@@ -41,6 +42,29 @@ type frame struct {
height uint16
}
+// EgressSlot은 실행 중인 파이프라인에 egress를 나중에 꽂거나 뗄 수 있게 하는
+// 홀더다. 명시적 송출 시작(#83)은 트랙 도착(파이프라인 기동) 이후에 오므로,
+// 파이프라인은 egress를 직접 들지 않고 이 슬롯을 통해서만 참조한다.
+type EgressSlot struct {
+ current atomic.Pointer[RTMPEgress]
+}
+
+func NewEgressSlot() *EgressSlot { return &EgressSlot{} }
+
+// Set은 활성 egress를 교체한다.
+func (s *EgressSlot) Set(egress *RTMPEgress) { s.current.Store(egress) }
+
+// Clear는 슬롯을 비운다. 이후 파이프라인 프레임은 egress로 가지 않는다.
+func (s *EgressSlot) Clear() { s.current.Store(nil) }
+
+// Load는 현재 활성 egress를 돌려준다(슬롯이 nil이거나 비었으면 nil).
+func (s *EgressSlot) Load() *RTMPEgress {
+ if s == nil {
+ return nil
+ }
+ return s.current.Load()
+}
+
type rtpSequenceObservation struct {
gap uint64
recovered bool
@@ -166,7 +190,7 @@ func RunTrack(
local *webrtc.TrackLocalStaticSample,
processor *Processor,
transcoder *FFmpegTranscoder,
- egress *RTMPEgress,
+ egress *EgressSlot,
registry *metrics.Registry,
mode config.PrivacyMode,
queueSize int,
@@ -182,7 +206,7 @@ func runTranscodedTrack(
local *webrtc.TrackLocalStaticSample,
processor *Processor,
transcoder *FFmpegTranscoder,
- egress *RTMPEgress,
+ egress *EgressSlot,
registry *metrics.Registry,
mode config.PrivacyMode,
queueSize int,
@@ -324,7 +348,7 @@ func processImages(
processor *Processor,
decoded <-chan frame,
processed chan<- frame,
- egress *RTMPEgress,
+ egress *EgressSlot,
registry *metrics.Registry,
mode config.PrivacyMode,
) {
@@ -348,8 +372,8 @@ func processImages(
registry.IncFrameProcessed(string(mode))
item.data = output
item.stageAt = time.Now()
- if egress != nil {
- egress.Enqueue(item)
+ if sink := egress.Load(); sink != nil {
+ sink.Enqueue(item)
}
select {
case processed <- item:
diff --git a/internal/server/server.go b/internal/server/server.go
index 3f2bf13..0b22ee1 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -21,6 +21,7 @@ import (
"inno-live-server/internal/metrics"
"inno-live-server/internal/origin"
"inno-live-server/internal/session"
+ "inno-live-server/internal/streaming"
)
const maxJSONBody = 1 << 20
@@ -33,6 +34,7 @@ type Server struct {
ai *ai.Pool
references *referenceStore
origins origin.Config
+ streaming map[auth.StreamingProvider]streaming.Provider
authenticateUser func(context.Context, string) (uuid.UUID, error)
handler http.Handler
}
@@ -45,6 +47,7 @@ func New(
aiPool *ai.Pool,
origins origin.Config,
requireUser func(http.Handler) http.Handler,
+ streamingProviders map[auth.StreamingProvider]streaming.Provider,
userAuthenticators ...func(context.Context, string) (uuid.UUID, error),
) *Server {
if requireUser == nil {
@@ -58,6 +61,7 @@ func New(
ai: aiPool,
references: newReferenceStore(cfg.ReferenceStorePath, cfg.AIMeImagePath != ""),
origins: origins,
+ streaming: streamingProviders,
}
if len(userAuthenticators) > 0 {
s.authenticateUser = userAuthenticators[0]
@@ -176,16 +180,72 @@ func (s *Server) handleDeleteSession(w http.ResponseWriter, _ *http.Request, liv
w.WriteHeader(http.StatusNoContent)
}
-func (s *Server) handleStartStream(w http.ResponseWriter, _ *http.Request, liveSession *session.Session) {
- response := liveSession.Response()
- if response.Media.RawVideoTrack == nil {
- writeError(w, apiError{Status: http.StatusConflict, Code: "conflict", Message: "Cannot start stream before a video track is available.", Details: map[string]any{"session_id": liveSession.ID}})
+// handleStartStream은 명시적 송출 시작(#83)이다: 요청 사용자의 연결된 플랫폼
+// 계정으로 방송을 준비(Prepare)하고, 세션의 처리 출력에 egress를 붙인다.
+func (s *Server) handleStartStream(w http.ResponseWriter, r *http.Request, liveSession *session.Session) {
+ request := struct {
+ Provider string `json:"provider"`
+ Title string `json:"title"`
+ Privacy string `json:"privacy"`
+ }{}
+ r.Body = http.MaxBytesReader(w, r.Body, maxJSONBody)
+ if err := decodeOptionalJSON(r.Body, &request); err != nil {
+ writeError(w, badRequest("Invalid stream start request.", map[string]any{"error": err.Error()}))
+ return
+ }
+ providerName := auth.StreamingProvider(strings.TrimSpace(request.Provider))
+ if providerName == "" {
+ providerName = auth.StreamingProviderYouTube
+ }
+ provider := s.streaming[providerName]
+ if provider == nil {
+ // 플랫폼 송출이 조립되지 않은 배포(자격증명 미설정·벤치)에서는 종전
+ // 계약(501)을 유지한다.
+ writeError(w, apiError{Status: http.StatusNotImplemented, Code: "not_supported", Message: "Streaming to this platform is not configured on the server.", Details: map[string]any{"provider": providerName}})
+ return
+ }
+ prepared, err := provider.Prepare(r.Context(), liveSession.UserID, streaming.PrepareOptions{
+ Title: request.Title,
+ Privacy: request.Privacy,
+ })
+ if err != nil {
+ 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, 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:
+ s.logger.Error("prepare platform broadcast failed", "session_id", liveSession.ID, "provider", providerName, "error", err)
+ writeError(w, apiError{Status: http.StatusBadGateway, Code: "streaming_prepare_failed", Message: "The streaming platform could not prepare the broadcast."})
+ }
return
}
- writeError(w, apiError{Status: http.StatusNotImplemented, Code: "not_supported", Message: "RTMP publishing is not supported by the media server."})
+ if _, err := s.sessions.StartStream(liveSession.ID, prepared.IngestURL); err != nil {
+ switch {
+ case errors.Is(err, session.ErrNoVideoTrack):
+ writeError(w, apiError{Status: http.StatusConflict, Code: "conflict", Message: "Cannot start stream before a video track is available.", Details: map[string]any{"session_id": liveSession.ID}})
+ case errors.Is(err, session.ErrStreamActive):
+ writeError(w, apiError{Status: http.StatusConflict, Code: "stream_already_active", Message: "The stream is already active.", Details: map[string]any{"session_id": liveSession.ID}})
+ default:
+ writeSessionError(w, err, liveSession.ID)
+ }
+ return
+ }
+ writeJSON(w, http.StatusOK, liveSession.Response())
}
+// handleStopStream은 egress만 종료한다(뷰어 송출·세션은 유지). 플랫폼 쪽
+// 방송 종료는 autoStop이 담당하므로(송출 중단 약 1분 후 반영) 플랫폼 API
+// 호출이 없다.
func (s *Server) handleStopStream(w http.ResponseWriter, _ *http.Request, liveSession *session.Session) {
+ if _, err := s.sessions.StopStream(liveSession.ID); err != nil {
+ if errors.Is(err, session.ErrStreamNotActive) {
+ writeError(w, apiError{Status: http.StatusConflict, Code: "stream_not_active", Message: "The stream is not active.", Details: map[string]any{"session_id": liveSession.ID}})
+ return
+ }
+ writeSessionError(w, err, liveSession.ID)
+ return
+ }
writeJSON(w, http.StatusOK, liveSession.Response().Stream)
}
diff --git a/internal/server/server_test.go b/internal/server/server_test.go
index 942fa98..170693c 100644
--- a/internal/server/server_test.go
+++ b/internal/server/server_test.go
@@ -623,7 +623,7 @@ func newTestApplicationWithUserMiddleware(t *testing.T, requireUser func(http.Ha
if err != nil {
t.Fatal(err)
}
- return New(cfg, logger, registry, manager, nil, origins, requireUser, authenticateUsers...), manager
+ return New(cfg, logger, registry, manager, nil, origins, requireUser, nil, authenticateUsers...), manager
}
// createTestSession creates a session and returns its response plus the
diff --git a/internal/server/session_auth_test.go b/internal/server/session_auth_test.go
index fd64b81..5d7fb9d 100644
--- a/internal/server/session_auth_test.go
+++ b/internal/server/session_auth_test.go
@@ -124,7 +124,7 @@ func TestSessionAuthDisabledBypass(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- httpServer := httptest.NewServer(New(cfg, logger, registry, manager, nil, origins, nil).Handler())
+ httpServer := httptest.NewServer(New(cfg, logger, registry, manager, nil, origins, nil, nil).Handler())
defer httpServer.Close()
created, _ := createTestSession(t, httpServer.URL, nil)
diff --git a/internal/server/static/client/app.js b/internal/server/static/client/app.js
index ed297c3..71317b8 100644
--- a/internal/server/static/client/app.js
+++ b/internal/server/static/client/app.js
@@ -108,6 +108,8 @@ function bindElements() {
"clearLogBtn",
"copyJsonBtn",
"sessionJson",
+ "connectYoutubeBtn",
+ "youtubeDetail",
]) {
els[id] = document.getElementById(id);
}
@@ -139,6 +141,7 @@ function bindEvents() {
els.signUpBtn.addEventListener("click", () => void requestSignup());
els.verifyBtn.addEventListener("click", () => void verifySignup());
els.signOutBtn.addEventListener("click", () => void signOut());
+ els.connectYoutubeBtn.addEventListener("click", () => void connectYoutube());
els.authPassword.addEventListener("keydown", (event) => {
if (event.key === "Enter") {
void signIn();
@@ -666,6 +669,11 @@ function renderAuth() {
els.verifyRow.hidden = !verifying;
els.verifyBtn.hidden = !verifying;
els.signOutBtn.hidden = !signedIn;
+ // YouTube 연결은 로그인(이메일)과 별개의 부가 기능이다 — 로그인 상태에서만 노출.
+ els.connectYoutubeBtn.hidden = !signedIn;
+ if (!signedIn) {
+ els.youtubeDetail.hidden = true;
+ }
els.authDetail.textContent = signedIn
? `${state.authEmail} 로 로그인됨. 세션 API를 사용할 수 있습니다.`
: verifying
@@ -674,6 +682,66 @@ function renderAuth() {
updateButtons();
}
+function setYoutubeDetail(text, isError) {
+ els.youtubeDetail.hidden = false;
+ els.youtubeDetail.textContent = text;
+ els.youtubeDetail.style.color = isError ? "var(--danger, #b00020)" : "";
+}
+
+// connectYoutube는 GIS 팝업으로 인가 코드를 받아 서버에 전달해 YouTube 계정을
+// 연결한다. 로그인 자체는 이메일 그대로이고, 이 팝업은 송출 대상 연결 전용이다.
+// 코드 교환·토큰 보관은 전부 서버 몫이라 브라우저에는 인가 코드만 스친다.
+async function connectYoutube() {
+ if (!state.accessToken) {
+ setYoutubeDetail("먼저 로그인하세요.", true);
+ return;
+ }
+ if (!window.google?.accounts?.oauth2) {
+ setYoutubeDetail("Google 스크립트를 아직 불러오지 못했습니다. 잠시 후 다시 시도하세요.", true);
+ return;
+ }
+ let config;
+ try {
+ config = await apiFetch("/auth/youtube/config");
+ } catch (error) {
+ setYoutubeDetail(`서버에서 YouTube 연동 설정을 받지 못했습니다: ${error.message}`, true);
+ return;
+ }
+ setYoutubeDetail("Google 팝업에서 계정을 선택하고 동의해 주세요...");
+ const codeClient = window.google.accounts.oauth2.initCodeClient({
+ client_id: config.web_client_id,
+ scope: config.scope,
+ ux_mode: "popup",
+ callback: (response) => {
+ if (!response.code) {
+ setYoutubeDetail("Google이 인가 코드를 돌려주지 않았습니다.", true);
+ return;
+ }
+ void (async () => {
+ try {
+ const result = await apiFetch("/auth/youtube/connect", {
+ method: "POST",
+ body: JSON.stringify({
+ server_auth_code: response.code,
+ code_source: "web_popup",
+ }),
+ });
+ const title = result?.channel?.title || result?.channel?.id || "알 수 없는 채널";
+ setYoutubeDetail(`YouTube 연결됨: ${title}`);
+ logEvent("ok", "YouTube account connected", result);
+ } catch (error) {
+ setYoutubeDetail(`연결 실패: ${error.message}`, true);
+ logEvent("error", "YouTube connect failed", { message: error.message });
+ }
+ })();
+ },
+ error_callback: (error) => {
+ setYoutubeDetail(`Google 팝업 오류: ${error?.type || JSON.stringify(error)}`, true);
+ },
+ });
+ codeClient.requestCode();
+}
+
async function createSessionOnly() {
await runBusy(async () => {
const session = await createSession();
diff --git a/internal/server/static/client/index.html b/internal/server/static/client/index.html
index 8cf054f..79f5eb5 100644
--- a/internal/server/static/client/index.html
+++ b/internal/server/static/client/index.html
@@ -41,6 +41,10 @@ WebRTC Client
로그인이 필요합니다. 계정이 없으면 이메일·비밀번호 입력 후 회원가입하세요.
+
+
+
+
+
+