Merge branch 'master' of https://github.com/AlexxIT/go2rtc into preload

pkg/h265/rtp.go

@@ -9,8 +9,8 @@ import (
 )
 
 func RTPDepay(codec *core.Codec, handler core.HandlerFunc) core.HandlerFunc {
-	//vps, sps, pps := GetParameterSet(codec.FmtpLine)
-	//ps := h264.EncodeAVC(vps, sps, pps)
+	vps, sps, pps := GetParameterSet(codec.FmtpLine)
+	ps := h264.JoinNALU(vps, sps, pps)
 
 	buf := make([]byte, 0, 512*1024) // 512K
 	var nuStart int
@@ -40,9 +40,9 @@ func RTPDepay(codec *core.Codec, handler core.HandlerFunc) core.HandlerFunc {
 				nuType = data[2] & 0x3F
 
 				// push PS data before keyframe
-				//if len(buf) == 0 && nuType >= 19 && nuType <= 21 {
-				//	buf = append(buf, ps...)
-				//}
+				if len(buf) == 0 && nuType >= 19 && nuType <= 21 {
+					buf = append(buf, ps...)
+				}
 
 				nuStart = len(buf)
 				buf = append(buf, 0, 0, 0, 0) // NAL unit size

pkg/ring/api.go

@@ -11,9 +11,13 @@ import (
 	"net/http"
 	"reflect"
 	"strings"
+	"sync"
 	"time"
 )
 
+var clientCache = map[string]*RingApi{}
+var cacheMutex sync.Mutex
+
 type RefreshTokenAuth struct {
 	RefreshToken string
 }
@@ -23,13 +27,11 @@ type EmailAuth struct {
 	Password string
 }
 
-// AuthConfig represents the decoded refresh token data
 type AuthConfig struct {
 	RT  string `json:"rt"`  // Refresh Token
 	HID string `json:"hid"` // Hardware ID
 }
 
-// AuthTokenResponse represents the response from the authentication endpoint
 type AuthTokenResponse struct {
 	AccessToken  string `json:"access_token"`
 	ExpiresIn    int    `json:"expires_in"`
@@ -46,41 +48,50 @@ type Auth2faResponse struct {
 	NextTimeInSecs   int    `json:"next_time_in_secs"`
 }
 
-// SocketTicketRequest represents the request to get a socket ticket
 type SocketTicketResponse struct {
 	Ticket            string `json:"ticket"`
 	ResponseTimestamp int64  `json:"response_timestamp"`
 }
 
-// RingRestClient handles authentication and requests to Ring API
-type RingRestClient struct {
+type SessionResponse struct {
+	Profile struct {
+		ID        int64  `json:"id"`
+		Email     string `json:"email"`
+		FirstName string `json:"first_name"`
+		LastName  string `json:"last_name"`
+	} `json:"profile"`
+}
+
+type RingApi struct {
 	httpClient     *http.Client
 	authConfig     *AuthConfig
 	hardwareID     string
 	authToken      *AuthTokenResponse
+	tokenExpiry    time.Time
 	Using2FA       bool
 	PromptFor2FA   string
 	RefreshToken   string
 	auth           interface{} // EmailAuth or RefreshTokenAuth
 	onTokenRefresh func(string)
+	authMutex      sync.Mutex
+	session        *SessionResponse
+	sessionExpiry  time.Time
+	sessionMutex   sync.Mutex
+	cacheKey       string
 }
 
-// CameraKind represents the different types of Ring cameras
 type CameraKind string
 
-// CameraData contains common fields for all camera types
 type CameraData struct {
-	ID          float64 `json:"id"`
-	Description string  `json:"description"`
-	DeviceID    string  `json:"device_id"`
-	Kind        string  `json:"kind"`
-	LocationID  string  `json:"location_id"`
+	ID          int    `json:"id"`
+	Description string `json:"description"`
+	DeviceID    string `json:"device_id"`
+	Kind        string `json:"kind"`
+	LocationID  string `json:"location_id"`
 }
 
-// RingDeviceType represents different types of Ring devices
 type RingDeviceType string
 
-// RingDevicesResponse represents the response from the Ring API
 type RingDevicesResponse struct {
 	Doorbots           []CameraData             `json:"doorbots"`
 	AuthorizedDoorbots []CameraData             `json:"authorized_doorbots"`
@@ -139,23 +150,49 @@ const (
 	apiVersion         = 11
 	defaultTimeout     = 20 * time.Second
 	maxRetries         = 3
+	sessionValidTime   = 12 * time.Hour
 )
 
-// NewRingRestClient creates a new Ring client instance
-func NewRingRestClient(auth interface{}, onTokenRefresh func(string)) (*RingRestClient, error) {
-	client := &RingRestClient{
-		httpClient:     &http.Client{Timeout: defaultTimeout},
-		onTokenRefresh: onTokenRefresh,
-		hardwareID:     generateHardwareID(),
-		auth:           auth,
-	}
+func NewRestClient(auth interface{}, onTokenRefresh func(string)) (*RingApi, error) {
+	var cacheKey string
 
+	// Create cache key based on auth data
 	switch a := auth.(type) {
 	case RefreshTokenAuth:
 		if a.RefreshToken == "" {
 			return nil, fmt.Errorf("refresh token is required")
 		}
+		cacheKey = "refresh:" + a.RefreshToken
+	case EmailAuth:
+		if a.Email == "" || a.Password == "" {
+			return nil, fmt.Errorf("email and password are required")
+		}
+		cacheKey = "email:" + a.Email + ":" + a.Password
+	default:
+		return nil, fmt.Errorf("invalid auth type")
+	}
 
+	cacheMutex.Lock()
+	defer cacheMutex.Unlock()
+
+	if cachedClient, ok := clientCache[cacheKey]; ok {
+		// Check if token is not nil and not expired
+		if cachedClient.authToken != nil && time.Now().Before(cachedClient.tokenExpiry) {
+			cachedClient.onTokenRefresh = onTokenRefresh
+			return cachedClient, nil
+		}
+	}
+
+	client := &RingApi{
+		httpClient:     &http.Client{Timeout: defaultTimeout},
+		onTokenRefresh: onTokenRefresh,
+		hardwareID:     generateHardwareID(),
+		auth:           auth,
+		cacheKey:       cacheKey,
+	}
+
+	switch a := auth.(type) {
+	case RefreshTokenAuth:
 		config, err := parseAuthConfig(a.RefreshToken)
 		if err != nil {
 			return nil, fmt.Errorf("failed to parse refresh token: %w", err)
@@ -164,160 +201,30 @@ func NewRingRestClient(auth interface{}, onTokenRefresh func(string)) (*RingRest
 		client.authConfig = config
 		client.hardwareID = config.HID
 		client.RefreshToken = a.RefreshToken
-	case EmailAuth:
-		if a.Email == "" || a.Password == "" {
-			return nil, fmt.Errorf("email and password are required")
-		}
-	default:
-		return nil, fmt.Errorf("invalid auth type")
 	}
 
+	clientCache[cacheKey] = client
+
 	return client, nil
 }
 
-// Request makes an authenticated request to the Ring API
-func (c *RingRestClient) Request(method, url string, body interface{}) ([]byte, error) {
-	// Ensure we have a valid auth token
-	if err := c.ensureAuth(); err != nil {
-		return nil, fmt.Errorf("authentication failed: %w", err)
-	}
-
-	var bodyReader io.Reader
-	if body != nil {
-		jsonBody, err := json.Marshal(body)
-		if err != nil {
-			return nil, fmt.Errorf("failed to marshal request body: %w", err)
-		}
-		bodyReader = bytes.NewReader(jsonBody)
-	}
-
-	// Create request
-	req, err := http.NewRequest(method, url, bodyReader)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create request: %w", err)
-	}
-
-	// Set headers
-	req.Header.Set("Authorization", "Bearer "+c.authToken.AccessToken)
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Accept", "application/json")
-	req.Header.Set("hardware_id", c.hardwareID)
-	req.Header.Set("User-Agent", "android:com.ringapp")
-
-	// Make request with retries
-	var resp *http.Response
-	var responseBody []byte
-
-	for attempt := 0; attempt <= maxRetries; attempt++ {
-		resp, err = c.httpClient.Do(req)
-		if err != nil {
-			if attempt == maxRetries {
-				return nil, fmt.Errorf("request failed after %d retries: %w", maxRetries, err)
-			}
-			time.Sleep(5 * time.Second)
-			continue
-		}
-		defer resp.Body.Close()
-
-		responseBody, err = io.ReadAll(resp.Body)
-		if err != nil {
-			return nil, fmt.Errorf("failed to read response body: %w", err)
-		}
-
-		// Handle 401 by refreshing auth and retrying
-		if resp.StatusCode == http.StatusUnauthorized {
-			c.authToken = nil // Force token refresh
-			if attempt == maxRetries {
-				return nil, fmt.Errorf("authentication failed after %d retries", maxRetries)
-			}
-			if err := c.ensureAuth(); err != nil {
-				return nil, fmt.Errorf("failed to refresh authentication: %w", err)
-			}
-			req.Header.Set("Authorization", "Bearer "+c.authToken.AccessToken)
-			continue
-		}
-
-		// Handle other error status codes
-		if resp.StatusCode >= 400 {
-			return nil, fmt.Errorf("request failed with status %d: %s", resp.StatusCode, string(responseBody))
-		}
-
-		break
-	}
-
-	return responseBody, nil
+func ClientAPI(path string) string {
+	return clientAPIBaseURL + path
 }
 
-// ensureAuth ensures we have a valid auth token
-func (c *RingRestClient) ensureAuth() error {
-	if c.authToken != nil {
-		return nil
-	}
-
-	var grantData = map[string]string{
-		"grant_type":    "refresh_token",
-		"refresh_token": c.authConfig.RT,
-	}
-
-	// Add common fields
-	grantData["client_id"] = "ring_official_android"
-	grantData["scope"] = "client"
-
-	// Make auth request
-	body, err := json.Marshal(grantData)
-	if err != nil {
-		return fmt.Errorf("failed to marshal auth request: %w", err)
-	}
-
-	req, err := http.NewRequest("POST", oauthURL, bytes.NewReader(body))
-	if err != nil {
-		return fmt.Errorf("failed to create auth request: %w", err)
-	}
-
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Accept", "application/json")
-	req.Header.Set("hardware_id", c.hardwareID)
-	req.Header.Set("User-Agent", "android:com.ringapp")
-	req.Header.Set("2fa-support", "true")
-
-	resp, err := c.httpClient.Do(req)
-	if err != nil {
-		return fmt.Errorf("auth request failed: %w", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode == http.StatusPreconditionFailed {
-		return fmt.Errorf("2FA required. Please see documentation for handling 2FA")
-	}
-
-	if resp.StatusCode != http.StatusOK {
-		body, _ := io.ReadAll(resp.Body)
-		return fmt.Errorf("auth request failed with status %d: %s", resp.StatusCode, string(body))
-	}
-
-	var authResp AuthTokenResponse
-	if err := json.NewDecoder(resp.Body).Decode(&authResp); err != nil {
-		return fmt.Errorf("failed to decode auth response: %w", err)
-	}
-
-	// Update auth config and refresh token
-	c.authToken = &authResp
-	c.authConfig = &AuthConfig{
-		RT:  authResp.RefreshToken,
-		HID: c.hardwareID,
-	}
-
-	// Encode and notify about new refresh token
-	if c.onTokenRefresh != nil {
-		newRefreshToken := encodeAuthConfig(c.authConfig)
-		c.onTokenRefresh(newRefreshToken)
-	}
-
-	return nil
+func DeviceAPI(path string) string {
+	return deviceAPIBaseURL + path
 }
 
-// getAuth makes an authentication request to the Ring API
-func (c *RingRestClient) GetAuth(twoFactorAuthCode string) (*AuthTokenResponse, error) {
+func CommandsAPI(path string) string {
+	return commandsAPIBaseURL + path
+}
+
+func AppAPI(path string) string {
+	return appAPIBaseURL + path
+}
+
+func (c *RingApi) GetAuth(twoFactorAuthCode string) (*AuthTokenResponse, error) {
 	var grantData map[string]string
 
 	if c.authConfig != nil && twoFactorAuthCode == "" {
@@ -404,60 +311,30 @@ func (c *RingRestClient) GetAuth(twoFactorAuthCode string) (*AuthTokenResponse,
 		return nil, fmt.Errorf("failed to decode auth response: %w", err)
 	}
 
+	// Refresh token and expiry
 	c.authToken = &authResp
 	c.authConfig = &AuthConfig{
 		RT:  authResp.RefreshToken,
 		HID: c.hardwareID,
 	}
+	// Set token expiry (1 minute before actual expiry)
+	expiresIn := time.Duration(authResp.ExpiresIn-60) * time.Second
+	c.tokenExpiry = time.Now().Add(expiresIn)
 
 	c.RefreshToken = encodeAuthConfig(c.authConfig)
 	if c.onTokenRefresh != nil {
 		c.onTokenRefresh(c.RefreshToken)
 	}
 
+	// Refresh the cached client
+	cacheMutex.Lock()
+	clientCache[c.cacheKey] = c
+	cacheMutex.Unlock()
+
 	return c.authToken, nil
 }
 
-// Helper functions for auth config encoding/decoding
-func parseAuthConfig(refreshToken string) (*AuthConfig, error) {
-	decoded, err := base64.StdEncoding.DecodeString(refreshToken)
-	if err != nil {
-		return nil, err
-	}
-
-	var config AuthConfig
-	if err := json.Unmarshal(decoded, &config); err != nil {
-		// Handle legacy format where refresh token is the raw token
-		return &AuthConfig{RT: refreshToken}, nil
-	}
-
-	return &config, nil
-}
-
-func encodeAuthConfig(config *AuthConfig) string {
-	jsonBytes, _ := json.Marshal(config)
-	return base64.StdEncoding.EncodeToString(jsonBytes)
-}
-
-// API URL helpers
-func ClientAPI(path string) string {
-	return clientAPIBaseURL + path
-}
-
-func DeviceAPI(path string) string {
-	return deviceAPIBaseURL + path
-}
-
-func CommandsAPI(path string) string {
-	return commandsAPIBaseURL + path
-}
-
-func AppAPI(path string) string {
-	return appAPIBaseURL + path
-}
-
-// FetchRingDevices gets all Ring devices and categorizes them
-func (c *RingRestClient) FetchRingDevices() (*RingDevicesResponse, error) {
+func (c *RingApi) FetchRingDevices() (*RingDevicesResponse, error) {
 	response, err := c.Request("GET", ClientAPI("ring_devices"), nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch ring devices: %w", err)
@@ -509,7 +386,7 @@ func (c *RingRestClient) FetchRingDevices() (*RingDevicesResponse, error) {
 	return &devices, nil
 }
 
-func (c *RingRestClient) GetSocketTicket() (*SocketTicketResponse, error) {
+func (c *RingApi) GetSocketTicket() (*SocketTicketResponse, error) {
 	response, err := c.Request("POST", AppAPI("clap/ticket/request/signalsocket"), nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch socket ticket: %w", err)
@@ -523,6 +400,286 @@ func (c *RingRestClient) GetSocketTicket() (*SocketTicketResponse, error) {
 	return &ticket, nil
 }
 
+func (c *RingApi) Request(method, url string, body interface{}) ([]byte, error) {
+	// Ensure we have a valid session
+	if err := c.ensureSession(); err != nil {
+		return nil, fmt.Errorf("session validation failed: %w", err)
+	}
+
+	var bodyReader io.Reader
+	if body != nil {
+		jsonBody, err := json.Marshal(body)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal request body: %w", err)
+		}
+		bodyReader = bytes.NewReader(jsonBody)
+	}
+
+	// Create request
+	req, err := http.NewRequest(method, url, bodyReader)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request: %w", err)
+	}
+
+	// Set headers
+	req.Header.Set("Authorization", "Bearer "+c.authToken.AccessToken)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("hardware_id", c.hardwareID)
+	req.Header.Set("User-Agent", "android:com.ringapp")
+
+	// Make request with retries
+	var resp *http.Response
+	var responseBody []byte
+
+	for attempt := 0; attempt <= maxRetries; attempt++ {
+		resp, err = c.httpClient.Do(req)
+		if err != nil {
+			if attempt == maxRetries {
+				return nil, fmt.Errorf("request failed after %d retries: %w", maxRetries, err)
+			}
+			time.Sleep(5 * time.Second)
+			continue
+		}
+		defer resp.Body.Close()
+
+		responseBody, err = io.ReadAll(resp.Body)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read response body: %w", err)
+		}
+
+		// Handle 401 by refreshing auth and retrying
+		if resp.StatusCode == http.StatusUnauthorized {
+			// Reset token to force refresh
+			c.authMutex.Lock()
+			c.authToken = nil
+			c.tokenExpiry = time.Time{} // Reset token expiry
+			c.authMutex.Unlock()
+
+			if attempt == maxRetries {
+				return nil, fmt.Errorf("authentication failed after %d retries", maxRetries)
+			}
+
+			// By 401 with Auth AND Session start over
+			c.sessionMutex.Lock()
+			c.session = nil
+			c.sessionExpiry = time.Time{} // Reset session expiry
+			c.sessionMutex.Unlock()
+
+			if err := c.ensureSession(); err != nil {
+				return nil, fmt.Errorf("failed to refresh session: %w", err)
+			}
+
+			req.Header.Set("Authorization", "Bearer "+c.authToken.AccessToken)
+			continue
+		}
+
+		// Handle 404 error with hardware_id reference - session issue
+		if resp.StatusCode == 404 && strings.Contains(url, clientAPIBaseURL) {
+			var errorBody map[string]interface{}
+			if err := json.Unmarshal(responseBody, &errorBody); err == nil {
+				if errorStr, ok := errorBody["error"].(string); ok && strings.Contains(errorStr, c.hardwareID) {
+					// Session with hardware_id not found, refresh session
+					c.sessionMutex.Lock()
+					c.session = nil
+					c.sessionExpiry = time.Time{} // Reset session expiry
+					c.sessionMutex.Unlock()
+
+					if attempt == maxRetries {
+						return nil, fmt.Errorf("session refresh failed after %d retries", maxRetries)
+					}
+
+					if err := c.ensureSession(); err != nil {
+						return nil, fmt.Errorf("failed to refresh session: %w", err)
+					}
+
+					continue
+				}
+			}
+		}
+
+		// Handle other error status codes
+		if resp.StatusCode >= 400 {
+			return nil, fmt.Errorf("request failed with status %d: %s", resp.StatusCode, string(responseBody))
+		}
+
+		break
+	}
+
+	return responseBody, nil
+}
+
+func (c *RingApi) ensureSession() error {
+	c.sessionMutex.Lock()
+	defer c.sessionMutex.Unlock()
+
+	// If session is still valid, use it
+	if c.session != nil && time.Now().Before(c.sessionExpiry) {
+		return nil
+	}
+
+	// Make sure we have a valid auth token
+	if err := c.ensureAuth(); err != nil {
+		return fmt.Errorf("authentication failed while creating session: %w", err)
+	}
+
+	sessionPayload := map[string]interface{}{
+		"device": map[string]interface{}{
+			"hardware_id": c.hardwareID,
+			"metadata": map[string]interface{}{
+				"api_version":  apiVersion,
+				"device_model": "ring-client-go",
+			},
+			"os": "android",
+		},
+	}
+
+	body, err := json.Marshal(sessionPayload)
+	if err != nil {
+		return fmt.Errorf("failed to marshal session request: %w", err)
+	}
+
+	req, err := http.NewRequest("POST", ClientAPI("session"), bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("Authorization", "Bearer "+c.authToken.AccessToken)
+	req.Header.Set("hardware_id", c.hardwareID)
+	req.Header.Set("User-Agent", "android:com.ringapp")
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		respBody, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("session request failed with status %d: %s", resp.StatusCode, string(respBody))
+	}
+
+	var sessionResp SessionResponse
+	if err := json.NewDecoder(resp.Body).Decode(&sessionResp); err != nil {
+		return fmt.Errorf("failed to decode session response: %w", err)
+	}
+
+	c.session = &sessionResp
+	c.sessionExpiry = time.Now().Add(sessionValidTime)
+
+	// Aktualisiere den gecachten Client
+	cacheMutex.Lock()
+	clientCache[c.cacheKey] = c
+	cacheMutex.Unlock()
+
+	return nil
+}
+
+func (c *RingApi) ensureAuth() error {
+	c.authMutex.Lock()
+	defer c.authMutex.Unlock()
+
+	// If token exists and is not expired, use it
+	if c.authToken != nil && time.Now().Before(c.tokenExpiry) {
+		return nil
+	}
+
+	var grantData = map[string]string{
+		"grant_type":    "refresh_token",
+		"refresh_token": c.authConfig.RT,
+	}
+
+	// Add common fields
+	grantData["client_id"] = "ring_official_android"
+	grantData["scope"] = "client"
+
+	// Make auth request
+	body, err := json.Marshal(grantData)
+	if err != nil {
+		return fmt.Errorf("failed to marshal auth request: %w", err)
+	}
+
+	req, err := http.NewRequest("POST", oauthURL, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("failed to create auth request: %w", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("hardware_id", c.hardwareID)
+	req.Header.Set("User-Agent", "android:com.ringapp")
+	req.Header.Set("2fa-support", "true")
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("auth request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusPreconditionFailed {
+		return fmt.Errorf("2FA required. Please see documentation for handling 2FA")
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("auth request failed with status %d: %s", resp.StatusCode, string(body))
+	}
+
+	var authResp AuthTokenResponse
+	if err := json.NewDecoder(resp.Body).Decode(&authResp); err != nil {
+		return fmt.Errorf("failed to decode auth response: %w", err)
+	}
+
+	// Update auth config and refresh token
+	c.authToken = &authResp
+	c.authConfig = &AuthConfig{
+		RT:  authResp.RefreshToken,
+		HID: c.hardwareID,
+	}
+
+	// Set token expiry (1 minute before actual expiry)
+	expiresIn := time.Duration(authResp.ExpiresIn-60) * time.Second
+	c.tokenExpiry = time.Now().Add(expiresIn)
+
+	// Encode and notify about new refresh token
+	if c.onTokenRefresh != nil {
+		newRefreshToken := encodeAuthConfig(c.authConfig)
+		c.onTokenRefresh(newRefreshToken)
+	}
+
+	// Refreshn the token in the client
+	c.RefreshToken = encodeAuthConfig(c.authConfig)
+
+	// Refresh the cached client
+	cacheMutex.Lock()
+	clientCache[c.cacheKey] = c
+	cacheMutex.Unlock()
+
+	return nil
+}
+
+func parseAuthConfig(refreshToken string) (*AuthConfig, error) {
+	decoded, err := base64.StdEncoding.DecodeString(refreshToken)
+	if err != nil {
+		return nil, err
+	}
+
+	var config AuthConfig
+	if err := json.Unmarshal(decoded, &config); err != nil {
+		// Handle legacy format where refresh token is the raw token
+		return &AuthConfig{RT: refreshToken}, nil
+	}
+
+	return &config, nil
+}
+
+func encodeAuthConfig(config *AuthConfig) string {
+	jsonBytes, _ := json.Marshal(config)
+	return base64.StdEncoding.EncodeToString(jsonBytes)
+}
+
 func generateHardwareID() string {
 	h := sha256.New()
 	h.Write([]byte("ring-client-go2rtc"))

pkg/ring/client.go

@@ -5,103 +5,25 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
-	"sync"
-	"time"
+	"strconv"
 
 	"github.com/AlexxIT/go2rtc/pkg/core"
 	"github.com/AlexxIT/go2rtc/pkg/webrtc"
 	"github.com/google/uuid"
-	"github.com/gorilla/websocket"
 	pion "github.com/pion/webrtc/v4"
 )
 
 type Client struct {
-	api       *RingRestClient
-	ws        *websocket.Conn
+	api       *RingApi
+	wsClient  *WSClient
 	prod      core.Producer
-	camera    *CameraData
+	cameraID  int
 	dialogID  string
-	sessionID string
-	wsMutex   sync.Mutex
-	done      chan struct{}
+	connected core.Waiter
+	closed    bool
 }
 
-type SessionBody struct {
-	DoorbotID int    `json:"doorbot_id"`
-	SessionID string `json:"session_id"`
-}
-
-type AnswerMessage struct {
-	Method string `json:"method"` // "sdp"
-	Body   struct {
-		SessionBody
-		SDP  string `json:"sdp"`
-		Type string `json:"type"` // "answer"
-	} `json:"body"`
-}
-
-type IceCandidateMessage struct {
-	Method string `json:"method"` // "ice"
-	Body   struct {
-		SessionBody
-		Ice        string `json:"ice"`
-		MLineIndex int    `json:"mlineindex"`
-	} `json:"body"`
-}
-
-type SessionMessage struct {
-	Method string      `json:"method"` // "session_created" or "session_started"
-	Body   SessionBody `json:"body"`
-}
-
-type PongMessage struct {
-	Method string      `json:"method"` // "pong"
-	Body   SessionBody `json:"body"`
-}
-
-type NotificationMessage struct {
-	Method string `json:"method"` // "notification"
-	Body   struct {
-		SessionBody
-		IsOK bool   `json:"is_ok"`
-		Text string `json:"text"`
-	} `json:"body"`
-}
-
-type StreamInfoMessage struct {
-	Method string `json:"method"` // "stream_info"
-	Body   struct {
-		SessionBody
-		Transcoding       bool   `json:"transcoding"`
-		TranscodingReason string `json:"transcoding_reason"`
-	} `json:"body"`
-}
-
-type CloseMessage struct {
-	Method string `json:"method"` // "close"
-	Body   struct {
-		SessionBody
-		Reason struct {
-			Code int    `json:"code"`
-			Text string `json:"text"`
-		} `json:"reason"`
-	} `json:"body"`
-}
-
-type BaseMessage struct {
-	Method string         `json:"method"`
-	Body   map[string]any `json:"body"`
-}
-
-// Close reason codes
-const (
-	CloseReasonNormalClose          = 0
-	CloseReasonAuthenticationFailed = 5
-	CloseReasonTimeout              = 6
-)
-
 func Dial(rawURL string) (*Client, error) {
-	// 1. Parse URL and validate basic params
 	u, err := url.Parse(rawURL)
 	if err != nil {
 		return nil, err
@@ -109,70 +31,42 @@ func Dial(rawURL string) (*Client, error) {
 
 	query := u.Query()
 	encodedToken := query.Get("refresh_token")
+	cameraID := query.Get("camera_id")
 	deviceID := query.Get("device_id")
 	_, isSnapshot := query["snapshot"]
 
-	if encodedToken == "" || deviceID == "" {
+	if encodedToken == "" || deviceID == "" || cameraID == "" {
 		return nil, errors.New("ring: wrong query")
 	}
 
-	// URL-decode the refresh token
+	client := &Client{
+		dialogID: uuid.NewString(),
+	}
+
+	client.cameraID, err = strconv.Atoi(cameraID)
+	if err != nil {
+		return nil, fmt.Errorf("ring: invalid camera_id: %w", err)
+	}
+
 	refreshToken, err := url.QueryUnescape(encodedToken)
 	if err != nil {
 		return nil, fmt.Errorf("ring: invalid refresh token encoding: %w", err)
 	}
 
-	// Initialize Ring API client
-	ringAPI, err := NewRingRestClient(RefreshTokenAuth{RefreshToken: refreshToken}, nil)
+	client.api, err = NewRestClient(RefreshTokenAuth{RefreshToken: refreshToken}, nil)
 	if err != nil {
 		return nil, err
 	}
 
-	// Get camera details
-	devices, err := ringAPI.FetchRingDevices()
-	if err != nil {
-		return nil, err
-	}
-
-	var camera *CameraData
-	for _, cam := range devices.AllCameras {
-		if fmt.Sprint(cam.DeviceID) == deviceID {
-			camera = &cam
-			break
-		}
-	}
-	if camera == nil {
-		return nil, errors.New("ring: camera not found")
-	}
-
-	// Create base client
-	client := &Client{
-		api:      ringAPI,
-		camera:   camera,
-		dialogID: uuid.NewString(),
-		done:     make(chan struct{}),
-	}
-
-	// Check if snapshot request
+	// Snapshot Flow
 	if isSnapshot {
-		client.prod = NewSnapshotProducer(ringAPI, camera)
+		client.prod = NewSnapshotProducer(client.api, client.cameraID)
 		return client, nil
 	}
 
-	// If not snapshot, continue with WebRTC setup
-	ticket, err := ringAPI.GetSocketTicket()
-	if err != nil {
-		return nil, err
-	}
-
-	// Create WebSocket connection
-	wsURL := fmt.Sprintf("wss://api.prod.signalling.ring.devices.a2z.com/ws?api_version=4.0&auth_type=ring_solutions&client_id=ring_site-%s&token=%s",
-		uuid.NewString(), url.QueryEscape(ticket.Ticket))
-
-	client.ws, _, err = websocket.DefaultDialer.Dial(wsURL, map[string][]string{
-		"User-Agent": {"android:com.ringapp"},
-	})
+	client.wsClient, err = StartWebsocket(client.cameraID, client.api)
 	if err != nil {
+		client.Stop()
 		return nil, err
 	}
 
@@ -196,13 +90,13 @@ func Dial(rawURL string) (*Client, error) {
 
 	api, err := webrtc.NewAPI()
 	if err != nil {
-		client.ws.Close()
+		client.Stop()
 		return nil, err
 	}
 
 	pc, err := api.NewPeerConnection(conf)
 	if err != nil {
-		client.ws.Close()
+		client.Stop()
 		return nil, err
 	}
 
@@ -212,16 +106,27 @@ func Dial(rawURL string) (*Client, error) {
 	// protect from blocking on errors
 	defer sendOffer.Done(nil)
 
-	// waiter will wait PC error or WS error or nil (connection OK)
-	var connState core.Waiter
-
 	prod := webrtc.NewConn(pc)
 	prod.FormatName = "ring/webrtc"
 	prod.Mode = core.ModeActiveProducer
 	prod.Protocol = "ws"
 	prod.URL = rawURL
 
-	client.prod = prod
+	client.wsClient.onMessage = func(msg WSMessage) {
+		client.onWSMessage(msg)
+	}
+
+	client.wsClient.onError = func(err error) {
+		// fmt.Printf("ring: error: %s\n", err.Error())
+		client.Stop()
+		client.connected.Done(err)
+	}
+
+	client.wsClient.onClose = func() {
+		// fmt.Println("ring: disconnect")
+		client.Stop()
+		client.connected.Done(errors.New("ring: disconnect"))
+	}
 
 	prod.Listen(func(msg any) {
 		switch msg := msg.(type) {
@@ -240,22 +145,28 @@ func Dial(rawURL string) (*Client, error) {
 				"mlineindex": iceCandidate.SDPMLineIndex,
 			}
 
-			if err = client.sendSessionMessage("ice", icePayload); err != nil {
-				connState.Done(err)
+			if err = client.wsClient.sendSessionMessage("ice", icePayload); err != nil {
+				client.connected.Done(err)
 				return
 			}
 
 		case pion.PeerConnectionState:
 			switch msg {
+			case pion.PeerConnectionStateNew:
+				break
 			case pion.PeerConnectionStateConnecting:
+				break
 			case pion.PeerConnectionStateConnected:
-				connState.Done(nil)
+				client.connected.Done(nil)
 			default:
-				connState.Done(errors.New("ring: " + msg.String()))
+				client.Stop()
+				client.connected.Done(errors.New("ring: " + msg.String()))
 			}
 		}
 	})
 
+	client.prod = prod
+
 	// Setup media configuration
 	medias := []*core.Media{
 		{
@@ -297,186 +208,103 @@ func Dial(rawURL string) (*Client, error) {
 		"sdp": offer,
 	}
 
-	if err = client.sendSessionMessage("live_view", offerPayload); err != nil {
+	if err = client.wsClient.sendSessionMessage("live_view", offerPayload); err != nil {
 		client.Stop()
 		return nil, err
 	}
 
 	sendOffer.Done(nil)
 
-	// Ring expects a ping message every 5 seconds
-	go client.startPingLoop(pc)
-	go client.startMessageLoop(&connState)
-
-	if err = connState.Wait(); err != nil {
+	if err = client.connected.Wait(); err != nil {
 		return nil, err
 	}
 
 	return client, nil
 }
 
-func (c *Client) startPingLoop(pc *pion.PeerConnection) {
-	ticker := time.NewTicker(5 * time.Second)
-	defer ticker.Stop()
+func (c *Client) onWSMessage(msg WSMessage) {
+	rawMsg, _ := json.Marshal(msg)
 
-	for {
-		select {
-		case <-c.done:
-			return
-		case <-ticker.C:
-			if pc.ConnectionState() == pion.PeerConnectionStateConnected {
-				if err := c.sendSessionMessage("ping", nil); err != nil {
-					return
-				}
-			}
+	// fmt.Printf("ring: onWSMessage: %s\n", string(rawMsg))
+
+	// check if "doorbot_id" is present
+	if _, ok := msg.Body["doorbot_id"]; !ok {
+		return
+	}
+
+	// check if the message is from the correct doorbot
+	doorbotID := msg.Body["doorbot_id"].(float64)
+	if int(doorbotID) != c.cameraID {
+		return
+	}
+
+	if msg.Method == "session_created" || msg.Method == "session_started" {
+		if _, ok := msg.Body["session_id"]; ok && c.wsClient.sessionID == "" {
+			c.wsClient.sessionID = msg.Body["session_id"].(string)
 		}
 	}
-}
 
-func (c *Client) startMessageLoop(connState *core.Waiter) {
-	var err error
-
-	// will be closed when conn will be closed
-	defer func() {
-		connState.Done(err)
-	}()
-
-	for {
-		select {
-		case <-c.done:
+	// check if the message is from the correct session
+	if _, ok := msg.Body["session_id"]; ok {
+		if msg.Body["session_id"].(string) != c.wsClient.sessionID {
 			return
-		default:
-			var res BaseMessage
-			if err = c.ws.ReadJSON(&res); err != nil {
-				select {
-				case <-c.done:
-					return
-				default:
-				}
+		}
+	}
 
+	switch msg.Method {
+	case "sdp":
+		if prod, ok := c.prod.(*webrtc.Conn); ok {
+			// Get answer
+			var msg AnswerMessage
+			if err := json.Unmarshal(rawMsg, &msg); err != nil {
 				c.Stop()
+				c.connected.Done(err)
 				return
 			}
 
-			// check if "doorbot_id" is present
-			if _, ok := res.Body["doorbot_id"]; !ok {
-				continue
-			}
-
-			// check if the message is from the correct doorbot
-			doorbotID := res.Body["doorbot_id"].(float64)
-			if doorbotID != float64(c.camera.ID) {
-				continue
-			}
-
-			// check if the message is from the correct session
-			if res.Method == "session_created" || res.Method == "session_started" {
-				if _, ok := res.Body["session_id"]; ok && c.sessionID == "" {
-					c.sessionID = res.Body["session_id"].(string)
-				}
-			}
-
-			if _, ok := res.Body["session_id"]; ok {
-				if res.Body["session_id"].(string) != c.sessionID {
-					continue
-				}
-			}
-
-			rawMsg, _ := json.Marshal(res)
-
-			switch res.Method {
-			case "sdp":
-				if prod, ok := c.prod.(*webrtc.Conn); ok {
-					// Get answer
-					var msg AnswerMessage
-					if err = json.Unmarshal(rawMsg, &msg); err != nil {
-						c.Stop()
-						return
-					}
-					if err = prod.SetAnswer(msg.Body.SDP); err != nil {
-						c.Stop()
-						return
-					}
-					if err = c.activateSession(); err != nil {
-						c.Stop()
-						return
-					}
-				}
-
-			case "ice":
-				if prod, ok := c.prod.(*webrtc.Conn); ok {
-					// Continue to receiving candidates
-					var msg IceCandidateMessage
-					if err = json.Unmarshal(rawMsg, &msg); err != nil {
-						break
-					}
-
-					// check for empty ICE candidate
-					if msg.Body.Ice == "" {
-						break
-					}
-
-					if err = prod.AddCandidate(msg.Body.Ice); err != nil {
-						c.Stop()
-						return
-					}
-				}
-
-			case "close":
+			if err := prod.SetAnswer(msg.Body.SDP); err != nil {
 				c.Stop()
+				c.connected.Done(err)
 				return
+			}
 
-			case "pong":
-				// Ignore
-				continue
+			if err := c.wsClient.activateSession(); err != nil {
+				c.Stop()
+				c.connected.Done(err)
+				return
+			}
+
+			prod.SDP = msg.Body.SDP
+		}
+
+	case "ice":
+		if prod, ok := c.prod.(*webrtc.Conn); ok {
+			var msg IceCandidateMessage
+			if err := json.Unmarshal(rawMsg, &msg); err != nil {
+				break
+			}
+
+			// Skip empty candidates
+			if msg.Body.Ice == "" {
+				break
+			}
+
+			if err := prod.AddCandidate(msg.Body.Ice); err != nil {
+				c.Stop()
+				c.connected.Done(err)
+				return
 			}
 		}
+
+	case "close":
+		c.Stop()
+		c.connected.Done(errors.New("ring: close"))
+
+	case "pong":
+		// Ignore
 	}
 }
 
-func (c *Client) activateSession() error {
-	if err := c.sendSessionMessage("activate_session", nil); err != nil {
-		return err
-	}
-
-	streamPayload := map[string]interface{}{
-		"audio_enabled": true,
-		"video_enabled": true,
-	}
-
-	if err := c.sendSessionMessage("stream_options", streamPayload); err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (c *Client) sendSessionMessage(method string, body map[string]interface{}) error {
-	c.wsMutex.Lock()
-	defer c.wsMutex.Unlock()
-
-	if body == nil {
-		body = make(map[string]interface{})
-	}
-
-	body["doorbot_id"] = c.camera.ID
-	if c.sessionID != "" {
-		body["session_id"] = c.sessionID
-	}
-
-	msg := map[string]interface{}{
-		"method":    method,
-		"dialog_id": c.dialogID,
-		"body":      body,
-	}
-
-	if err := c.ws.WriteJSON(msg); err != nil {
-		return err
-	}
-
-	return nil
-}
-
 func (c *Client) GetMedias() []*core.Media {
 	return c.prod.GetMedias()
 }
@@ -492,7 +320,7 @@ func (c *Client) AddTrack(media *core.Media, codec *core.Codec, track *core.Rece
 			speakerPayload := map[string]interface{}{
 				"stealth_mode": false,
 			}
-			_ = c.sendSessionMessage("camera_options", speakerPayload)
+			_ = c.wsClient.sendSessionMessage("camera_options", speakerPayload)
 		}
 		return webrtcProd.AddTrack(media, codec, track)
 	}
@@ -505,37 +333,23 @@ func (c *Client) Start() error {
 }
 
 func (c *Client) Stop() error {
-	select {
-	case <-c.done:
+	if c.closed {
 		return nil
-	default:
-		close(c.done)
 	}
 
+	c.closed = true
+
 	if c.prod != nil {
 		_ = c.prod.Stop()
 	}
 
-	if c.ws != nil {
-		closePayload := map[string]interface{}{
-			"reason": map[string]interface{}{
-				"code": CloseReasonNormalClose,
-				"text": "",
-			},
-		}
-
-		_ = c.sendSessionMessage("close", closePayload)
-		_ = c.ws.Close()
-		c.ws = nil
+	if c.wsClient != nil {
+		_ = c.wsClient.Close()
 	}
 
 	return nil
 }
 
 func (c *Client) MarshalJSON() ([]byte, error) {
-	if webrtcProd, ok := c.prod.(*webrtc.Conn); ok {
-		return webrtcProd.MarshalJSON()
-	}
-
 	return json.Marshal(c.prod)
 }

pkg/ring/snapshot.go

@@ -10,11 +10,11 @@ import (
 type SnapshotProducer struct {
 	core.Connection
 
-	client *RingRestClient
-	camera *CameraData
+	client   *RingApi
+	cameraID int
 }
 
-func NewSnapshotProducer(client *RingRestClient, camera *CameraData) *SnapshotProducer {
+func NewSnapshotProducer(client *RingApi, cameraID int) *SnapshotProducer {
 	return &SnapshotProducer{
 		Connection: core.Connection{
 			ID:         core.NewID(),
@@ -35,14 +35,13 @@ func NewSnapshotProducer(client *RingRestClient, camera *CameraData) *SnapshotPr
 				},
 			},
 		},
-		client: client,
-		camera: camera,
+		client:   client,
+		cameraID: cameraID,
 	}
 }
 
 func (p *SnapshotProducer) Start() error {
-	// Fetch snapshot
-	response, err := p.client.Request("GET", fmt.Sprintf("https://app-snaps.ring.com/snapshots/next/%d", int(p.camera.ID)), nil)
+	response, err := p.client.Request("GET", fmt.Sprintf("https://app-snaps.ring.com/snapshots/next/%d", p.cameraID), nil)
 	if err != nil {
 		return err
 	}

pkg/ring/ws.go

@@ -0,0 +1,265 @@
+package ring
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/gorilla/websocket"
+)
+
+type SessionBody struct {
+	DoorbotID int    `json:"doorbot_id"`
+	SessionID string `json:"session_id"`
+}
+
+type AnswerMessage struct {
+	Method string `json:"method"` // "sdp"
+	Body   struct {
+		SessionBody
+		SDP  string `json:"sdp"`
+		Type string `json:"type"` // "answer"
+	} `json:"body"`
+}
+
+type IceCandidateMessage struct {
+	Method string `json:"method"` // "ice"
+	Body   struct {
+		SessionBody
+		Ice        string `json:"ice"`
+		MLineIndex int    `json:"mlineindex"`
+	} `json:"body"`
+}
+
+type SessionMessage struct {
+	Method string      `json:"method"` // "session_created" or "session_started"
+	Body   SessionBody `json:"body"`
+}
+
+type PongMessage struct {
+	Method string      `json:"method"` // "pong"
+	Body   SessionBody `json:"body"`
+}
+
+type NotificationMessage struct {
+	Method string `json:"method"` // "notification"
+	Body   struct {
+		SessionBody
+		IsOK bool   `json:"is_ok"`
+		Text string `json:"text"`
+	} `json:"body"`
+}
+
+type StreamInfoMessage struct {
+	Method string `json:"method"` // "stream_info"
+	Body   struct {
+		SessionBody
+		Transcoding       bool   `json:"transcoding"`
+		TranscodingReason string `json:"transcoding_reason"`
+	} `json:"body"`
+}
+
+type CloseRequest struct {
+	Method string `json:"method"` // "close"
+	Body   struct {
+		SessionBody
+		Reason struct {
+			Code int    `json:"code"`
+			Text string `json:"text"`
+		} `json:"reason"`
+	} `json:"body"`
+}
+
+type WSMessage struct {
+	Method string         `json:"method"`
+	Body   map[string]any `json:"body"`
+}
+
+type WSClient struct {
+	ws        *websocket.Conn
+	api       *RingApi
+	wsMutex   sync.Mutex
+	cameraID  int
+	dialogID  string
+	sessionID string
+
+	onMessage func(msg WSMessage)
+	onError   func(err error)
+	onClose   func()
+
+	closed chan struct{}
+}
+
+const (
+	CloseReasonNormalClose          = 0
+	CloseReasonAuthenticationFailed = 5
+	CloseReasonTimeout              = 6
+)
+
+func StartWebsocket(cameraID int, api *RingApi) (*WSClient, error) {
+	client := &WSClient{
+		api:      api,
+		cameraID: cameraID,
+		dialogID: uuid.NewString(),
+		closed:   make(chan struct{}),
+	}
+
+	ticket, err := client.api.GetSocketTicket()
+	if err != nil {
+		return nil, err
+	}
+
+	url := fmt.Sprintf("wss://api.prod.signalling.ring.devices.a2z.com/ws?api_version=4.0&auth_type=ring_solutions&client_id=ring_site-%s&token=%s",
+		uuid.NewString(), url.QueryEscape(ticket.Ticket))
+
+	httpHeader := http.Header{}
+	httpHeader.Set("User-Agent", "android:com.ringapp")
+
+	client.ws, _, err = websocket.DefaultDialer.Dial(url, httpHeader)
+	if err != nil {
+		return nil, err
+	}
+
+	client.ws.SetCloseHandler(func(code int, text string) error {
+		client.onWsClose()
+		return nil
+	})
+
+	go client.startPingLoop()
+	go client.startMessageLoop()
+
+	return client, nil
+}
+
+func (c *WSClient) Close() error {
+	select {
+	case <-c.closed:
+		return nil
+	default:
+		close(c.closed)
+	}
+
+	closePayload := map[string]interface{}{
+		"reason": map[string]interface{}{
+			"code": CloseReasonNormalClose,
+			"text": "",
+		},
+	}
+
+	_ = c.sendSessionMessage("close", closePayload)
+
+	return c.ws.Close()
+}
+
+func (c *WSClient) startPingLoop() {
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-c.closed:
+			return
+		case <-ticker.C:
+			if err := c.sendSessionMessage("ping", nil); err != nil {
+				return
+			}
+		}
+	}
+}
+
+func (c *WSClient) startMessageLoop() {
+	for {
+		select {
+		case <-c.closed:
+			return
+		default:
+			var res WSMessage
+			if err := c.ws.ReadJSON(&res); err != nil {
+				select {
+				case <-c.closed:
+					// Ignore error if closed
+				default:
+					c.onWsError(err)
+				}
+
+				return
+			}
+
+			c.onWsMessage(res)
+		}
+	}
+}
+
+func (c *WSClient) activateSession() error {
+	if err := c.sendSessionMessage("activate_session", nil); err != nil {
+		return err
+	}
+
+	streamPayload := map[string]interface{}{
+		"audio_enabled": true,
+		"video_enabled": true,
+	}
+
+	if err := c.sendSessionMessage("stream_options", streamPayload); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *WSClient) sendSessionMessage(method string, payload map[string]interface{}) error {
+	select {
+	case <-c.closed:
+		return nil
+	default:
+		// continue
+	}
+
+	c.wsMutex.Lock()
+	defer c.wsMutex.Unlock()
+
+	if payload == nil {
+		payload = make(map[string]interface{})
+	}
+
+	payload["doorbot_id"] = c.cameraID
+	if c.sessionID != "" {
+		payload["session_id"] = c.sessionID
+	}
+
+	msg := map[string]interface{}{
+		"method":    method,
+		"dialog_id": c.dialogID,
+		"body":      payload,
+	}
+
+	// rawMsg, _ := json.Marshal(msg)
+	// fmt.Printf("ring: sendSessionMessage: %s: %s\n", method, string(rawMsg))
+
+	if err := c.ws.WriteJSON(msg); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (c *WSClient) onWsMessage(msg WSMessage) {
+	if c.onMessage != nil {
+		c.onMessage(msg)
+	}
+}
+
+func (c *WSClient) onWsError(err error) {
+	if c.onError != nil {
+		c.onError(err)
+	}
+}
+
+func (c *WSClient) onWsClose() {
+	if c.onClose != nil {
+		c.onClose()
+	}
+}