diff --git a/.gitignore b/.gitignore index 1e74a6c..d0b6a3c 100644 --- a/.gitignore +++ b/.gitignore @@ -246,6 +246,6 @@ go.work.sum build/ GEMINI.md code/package-lock.json -olcrtc !cmd/olcrtc/ !cmd/olcrtc/main_test.go +!pkg/ diff --git a/pkg/olcrtc/olcrtc.go b/pkg/olcrtc/olcrtc.go new file mode 100644 index 0000000..20099f1 --- /dev/null +++ b/pkg/olcrtc/olcrtc.go @@ -0,0 +1,219 @@ +// Package olcrtc exposes olcrtc as an embeddable Go library. +// +// Typical usage (direct engine, no service-specific auth): +// +// sess, err := olcrtc.New(ctx, olcrtc.Config{ +// Engine: "livekit", +// URL: "wss://sfu.example/", +// Token: "", +// }) +// +// Typical usage (built-in auth provider): +// +// sess, err := olcrtc.New(ctx, olcrtc.Config{ +// Auth: "telemost", +// RoomID: "", +// }) +// +// In both cases the caller must import the engine and (optionally) auth +// packages it needs via blank imports so their init() functions run: +// +// import ( +// _ "github.com/openlibrecommunity/olcrtc/internal/engine/livekit" +// _ "github.com/openlibrecommunity/olcrtc/internal/auth/telemost" +// ) +// +// Or use [RegisterDefaults] to pull in all built-in implementations at once. +package olcrtc + +import ( + "context" + "errors" + "fmt" + + "github.com/openlibrecommunity/olcrtc/internal/auth" + "github.com/openlibrecommunity/olcrtc/internal/carrier/builtin" + "github.com/openlibrecommunity/olcrtc/internal/engine" +) + +var ( + // ErrAuthOrEngineRequired is returned when neither auth nor engine+URL are supplied. + ErrAuthOrEngineRequired = errors.New("olcrtc: supply either Auth or Engine+URL") + // ErrURLRequired is returned when direct mode is used without a URL. + ErrURLRequired = errors.New("olcrtc: URL required when using direct engine mode") + // ErrTokenRequired is returned when direct mode is used without a token. + ErrTokenRequired = errors.New("olcrtc: Token required when using direct engine mode") +) + +// Config is the input to [New]. +type Config struct { + // --- built-in auth mode --- + // Auth is the name of a registered auth provider ("telemost", "jazz", "wbstream"). + // When set, RoomID is forwarded to the provider as the room reference. + Auth string + RoomID string + + // --- direct engine mode (Auth == "") --- + // Engine selects the SFU protocol ("livekit", "goolom", "salutejazz"). + // Defaults to "livekit" when Auth is empty. + Engine string + URL string + Token string + + // --- common --- + // Name is the display name used when joining the room. + Name string + // DNSServer is an optional custom DNS resolver (e.g. "1.1.1.1:53"). + DNSServer string + // ProxyAddr / ProxyPort configure an outbound SOCKS5 proxy. + ProxyAddr string + ProxyPort int + // OnData, when set, receives incoming data-channel bytes. If nil the + // session operates in video-track / media-only mode. + OnData func([]byte) +} + +// Session is the library handle returned by [New]. +// Connect must be called before Send. Close releases all resources. +type Session struct { + inner engine.Session + // refresh is stored so it survives reconnects via the engine's Refresh hook. + authProvider auth.Provider + authCfg auth.Config +} + +// RegisterDefaults registers all built-in engines and auth providers. +// Call once at program start if you want the full set without manual blank +// imports. Safe to call multiple times. +func RegisterDefaults() { + builtin.Register() +} + +// New creates a Session from cfg. The session is not connected yet; call +// [Session.Connect] when ready. +func New(ctx context.Context, cfg Config) (*Session, error) { + if cfg.Auth != "" { + return newWithAuth(ctx, cfg) + } + return newDirect(ctx, cfg) +} + +func newWithAuth(ctx context.Context, cfg Config) (*Session, error) { + p, err := auth.Get(cfg.Auth) + if err != nil { + return nil, fmt.Errorf("olcrtc: auth provider %q not registered: %w", cfg.Auth, err) + } + + authCfg := auth.Config{ + RoomURL: cfg.RoomID, + Name: cfg.Name, + DNSServer: cfg.DNSServer, + ProxyAddr: cfg.ProxyAddr, + ProxyPort: cfg.ProxyPort, + } + + creds, err := p.Issue(ctx, authCfg) + if err != nil { + return nil, fmt.Errorf("olcrtc: auth issue: %w", err) + } + + engineName := p.Engine() + sess, err := engine.New(ctx, engineName, engine.Config{ + URL: creds.URL, + Token: creds.Token, + Name: cfg.Name, + Extra: creds.Extra, + OnData: cfg.OnData, + DNSServer: cfg.DNSServer, + ProxyAddr: cfg.ProxyAddr, + ProxyPort: cfg.ProxyPort, + Refresh: func(rCtx context.Context) (engine.Credentials, error) { + fresh, freshErr := p.Issue(rCtx, authCfg) + if freshErr != nil { + return engine.Credentials{}, fmt.Errorf("olcrtc: auth refresh: %w", freshErr) + } + return engine.Credentials{URL: fresh.URL, Token: fresh.Token, Extra: fresh.Extra}, nil + }, + }) + if err != nil { + return nil, fmt.Errorf("olcrtc: engine %q: %w", engineName, err) + } + + return &Session{inner: sess, authProvider: p, authCfg: authCfg}, nil +} + +func newDirect(ctx context.Context, cfg Config) (*Session, error) { + if cfg.URL == "" { + return nil, ErrURLRequired + } + if cfg.Token == "" { + return nil, ErrTokenRequired + } + + engineName := cfg.Engine + if engineName == "" { + engineName = "livekit" + } + + sess, err := engine.New(ctx, engineName, engine.Config{ + URL: cfg.URL, + Token: cfg.Token, + Name: cfg.Name, + OnData: cfg.OnData, + DNSServer: cfg.DNSServer, + ProxyAddr: cfg.ProxyAddr, + ProxyPort: cfg.ProxyPort, + }) + if err != nil { + return nil, fmt.Errorf("olcrtc: engine %q: %w", engineName, err) + } + + return &Session{inner: sess}, nil +} + +// Connect establishes the WebRTC connection. Blocks until the data channel (or +// media) is ready, or ctx is cancelled. +func (s *Session) Connect(ctx context.Context) error { + if err := s.inner.Connect(ctx); err != nil { + return fmt.Errorf("connect: %w", err) + } + return nil +} + +// Send queues data for transmission over the data channel. +func (s *Session) Send(data []byte) error { + if err := s.inner.Send(data); err != nil { + return fmt.Errorf("send: %w", err) + } + return nil +} + +// Close tears down the session and releases all resources. +func (s *Session) Close() error { + if err := s.inner.Close(); err != nil { + return fmt.Errorf("close: %w", err) + } + return nil +} + +// WatchConnection monitors the connection and handles reconnects. Run in a +// goroutine alongside Connect. +func (s *Session) WatchConnection(ctx context.Context) { + s.inner.WatchConnection(ctx) +} + +// CanSend reports whether the session is ready to accept outgoing data. +func (s *Session) CanSend() bool { + return s.inner.CanSend() +} + +// SetEndedCallback registers a function called when the session ends +// permanently (after reconnect exhaustion or explicit close). +func (s *Session) SetEndedCallback(cb func(reason string)) { + s.inner.SetEndedCallback(cb) +} + +// SetShouldReconnect controls whether automatic reconnection is attempted. +func (s *Session) SetShouldReconnect(fn func() bool) { + s.inner.SetShouldReconnect(fn) +} diff --git a/pkg/olcrtc/olcrtc_test.go b/pkg/olcrtc/olcrtc_test.go new file mode 100644 index 0000000..a99b699 --- /dev/null +++ b/pkg/olcrtc/olcrtc_test.go @@ -0,0 +1,160 @@ +package olcrtc_test + +import ( + "context" + "errors" + "testing" + + "github.com/openlibrecommunity/olcrtc/internal/auth" + "github.com/openlibrecommunity/olcrtc/internal/engine" + "github.com/openlibrecommunity/olcrtc/pkg/olcrtc" + "github.com/pion/webrtc/v4" +) + +const ( + stubToken = "tok" + stubURL = "wss://x/" +) + +// --- stub engine --- + +type stubSession struct{ connected bool } + +func (s *stubSession) Connect(_ context.Context) error { s.connected = true; return nil } +func (s *stubSession) Send(_ []byte) error { return nil } +func (s *stubSession) Close() error { return nil } +func (s *stubSession) SetReconnectCallback(_ func(*webrtc.DataChannel)) {} +func (s *stubSession) SetShouldReconnect(_ func() bool) {} +func (s *stubSession) SetEndedCallback(_ func(string)) {} +func (s *stubSession) WatchConnection(_ context.Context) {} +func (s *stubSession) CanSend() bool { return s.connected } +func (s *stubSession) GetSendQueue() chan []byte { return nil } +func (s *stubSession) GetBufferedAmount() uint64 { return 0 } +func (s *stubSession) Capabilities() engine.Capabilities { return engine.Capabilities{ByteStream: true} } + +// Compile-time check: stubSession must satisfy engine.Session. +var _ engine.Session = (*stubSession)(nil) + +func registerStubEngine(t *testing.T, name string) { + t.Helper() + engine.Register(name, func(_ context.Context, _ engine.Config) (engine.Session, error) { + return &stubSession{}, nil + }) + t.Cleanup(func() { + // Re-register a no-op so subsequent tests don't break. + engine.Register(name, func(_ context.Context, _ engine.Config) (engine.Session, error) { + return &stubSession{}, nil + }) + }) +} + +// --- stub auth --- + +type stubAuth struct{ engineName string } + +func (a stubAuth) Engine() string { return a.engineName } +func (a stubAuth) Issue(_ context.Context, cfg auth.Config) (auth.Credentials, error) { + if cfg.RoomURL == "" { + return auth.Credentials{}, auth.ErrRoomIDRequired + } + return auth.Credentials{URL: "wss://stub/", Token: stubToken}, nil +} + +func registerStubAuth(t *testing.T, name, engineName string) { + t.Helper() + auth.Register(name, stubAuth{engineName: engineName}) +} + +// --- tests --- + +func TestNewDirect_MissingURL(t *testing.T) { + _, err := olcrtc.New(context.Background(), olcrtc.Config{Token: "tok"}) + if !errors.Is(err, olcrtc.ErrURLRequired) { + t.Fatalf("New(no url) = %v, want ErrURLRequired", err) + } +} + +func TestNewDirect_MissingToken(t *testing.T) { + _, err := olcrtc.New(context.Background(), olcrtc.Config{URL: stubURL}) + if !errors.Is(err, olcrtc.ErrTokenRequired) { + t.Fatalf("New(no token) = %v, want ErrTokenRequired", err) + } +} + +func TestNewDirect_UnknownEngine(t *testing.T) { + _, err := olcrtc.New(context.Background(), olcrtc.Config{ + Engine: "no-such-engine", + URL: stubURL, + Token: stubToken, + }) + if err == nil { + t.Fatal("New(bad engine) error = nil") + } +} + +func TestNewDirect_OK(t *testing.T) { + registerStubEngine(t, "stub-direct") + + sess, err := olcrtc.New(context.Background(), olcrtc.Config{ + Engine: "stub-direct", + URL: stubURL, + Token: stubToken, + }) + if err != nil { + t.Fatalf("New() error = %v", err) + } + if err := sess.Connect(context.Background()); err != nil { + t.Fatalf("Connect() error = %v", err) + } + if !sess.CanSend() { + t.Fatal("CanSend() = false after connect") + } + if err := sess.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } +} + +func TestNewAuth_UnknownProvider(t *testing.T) { + _, err := olcrtc.New(context.Background(), olcrtc.Config{ + Auth: "no-such-auth", + RoomID: "room", + }) + if err == nil { + t.Fatal("New(bad auth) error = nil") + } +} + +func TestNewAuth_MissingRoomID(t *testing.T) { + registerStubEngine(t, "stub-auth-engine") + registerStubAuth(t, "stub-auth-noroomid", "stub-auth-engine") + + _, err := olcrtc.New(context.Background(), olcrtc.Config{ + Auth: "stub-auth-noroomid", + // RoomID intentionally empty + }) + if err == nil { + t.Fatal("New(auth, no room) error = nil") + } +} + +func TestNewAuth_OK(t *testing.T) { + registerStubEngine(t, "stub-auth-ok-engine") + registerStubAuth(t, "stub-auth-ok", "stub-auth-ok-engine") + + sess, err := olcrtc.New(context.Background(), olcrtc.Config{ + Auth: "stub-auth-ok", + RoomID: "some-room", + }) + if err != nil { + t.Fatalf("New(auth) error = %v", err) + } + if err := sess.Connect(context.Background()); err != nil { + t.Fatalf("Connect() error = %v", err) + } + _ = sess.Close() +} + +func TestRegisterDefaults_Idempotent(_ *testing.T) { + olcrtc.RegisterDefaults() + olcrtc.RegisterDefaults() +}