openvibers/openvibe.link Commits

Host pool party desktops from the Link service (#150)

OP OpenVibers committed 3 days ago · 47652791db · 8 files, +213 −1

* Host pool party desktops from the Link service

* Use the configured party URL when the pool starts a host

Changes

ddiff --git a/docs/party-desktop.md b/docs/party-desktop.md
iindex d217f67..1fe0655 100644
---- a/docs/party-desktop.md
++++ b/docs/party-desktop.md
@@@ -13,3 +13,25 @@ When the Link service is running, the command starts hosting in the background.
The party's existing remote decides who can drive. Members use **Ask for it**, **Pass it**, and **Take it back** on openvibe.party. The shared browser is framed in the party page, and members must sign in to OpenVibe before they can watch. **Listen** uses the same browser for Spotify or YouTube Music.
For operators: Chat requires `TURN_URL=turns:<host>:443` with `TURN_AUTH_SECRET`, and the TURN service must accept TLS on port 443. Chat's `CHAT_PARTY_CONNECT_SECRET` and Connect's `CONNECT_PARTY_SHARED_SECRET` must be the same random value of at least 32 characters. The Link needs Network-issued node tokens for `openvibe.chat` and `openvibe.link`.
++
++## OpenVibe's own computer pool
++
++A machine can also serve parties from OpenVibe's own pool, with no person at it to run a command. The owner registers
++the machine on [openvibe.party/pool](https://openvibe.party/pool) with its Link principal (`lnk_…`), a name, and how
++many desktops it may host at once. On the machine, the Link service offers itself with:
++
++```json
++{
++ "party_pool": { "enabled": true }
++}
++```
++
++It then asks Chat every five seconds, with the machine's own `openvibe.chat` node token, what the pool is holding for
++it (`GET /api/chat/parties/desktop/assignment`). An answer starts the same host path as `openvibe-link party host`: the
++machine claims the party with its node token, opens its Connect share, reports ready, and renews by polling the
++party's `access` while it hosts. The poll is also how the pool sees the machine is alive. `party_pool.party_url` is
++only for a test copy of the site (an https origin, or loopback http).
++
++The owner's switch on the pool panel is off by default; turning it off stops new parties being placed there. A hold the
++machine never claims, or stops renewing, is released after the idle window, and the party page says the computer was
++released.
ddiff --git a/internal/agent/agent.go b/internal/agent/agent.go
iindex 6310dd1..9ac06a9 100644
---- a/internal/agent/agent.go
++++ b/internal/agent/agent.go
@@@ -339,6 +339,11 @@ func (n *Agent) Run(parent context.Context) error {
wg.Add(1)
go func() { defer wg.Done(); n.fleet.Run(ctx) }()
}
++ // OpenVibe's own computer pool (docs/party-desktop.md): a machine the owner registered asks Chat what to host.
++ if n.opt.Config.PartyPool.Enabled {
++ wg.Add(1)
++ go func() { defer wg.Done(); n.pollPartyPool(ctx) }()
++ }
if n.link != nil {
wg.Add(1)
go func() {
ddiff --git a/internal/agent/party.go b/internal/agent/party.go
iindex ad84459..9eaea6c 100644
---- a/internal/agent/party.go
++++ b/internal/agent/party.go
@@@ -38,7 +38,11 @@ func (n *Agent) startParty(id string) error {
chatTokens := link.NewTokenSourceFor(n.opt.Creds, nil, "openvibe-link/"+n.opt.Version, "openvibe.chat")
connectTokens := link.NewTokenSourceFor(n.opt.Creds, nil, "openvibe-link/"+n.opt.Version, link.ConnectAudience)
-- client := party.Client{PartyURL: "https://openvibe.party", ConnectServer: connect.DefaultServer, PartyID: id,
++ partyURL := n.opt.Config.PartyPool.PartyURL
++ if partyURL == "" {
++ partyURL = "https://openvibe.party"
++ }
++ client := party.Client{PartyURL: partyURL, ConnectServer: connect.DefaultServer, PartyID: id,
ChatToken: func(ctx context.Context) (string, error) {
t, err := chatTokens.Token(ctx)
if err != nil {
ddiff --git a/internal/agent/party_pool.go b/internal/agent/party_pool.go
nnew file mode 100644
iindex 0000000..e02b37c
---- /dev/null
++++ b/internal/agent/party_pool.go
@@@ -0,0 +1,76 @@
++package agent
++
++import (
++ "context"
++ "time"
++
++ "github.com/OpenVibers/OpenVibe.Link/internal/party"
++)
++
++// partyPoolInterval is how often a pool machine asks Chat what to host. The pool sweeps idle holds every minute, so
++// five seconds is prompt without a busy poll; Chat answers null for a machine it is not holding a party for.
++const partyPoolInterval = 5 * time.Second
++
++// pollPartyPool offers this machine to OpenVibe's own computer pool (openvibe.party/pool): the owner registered its
++// principal there, and the Link asks Chat what is held for it. It starts the same host path as
++// `openvibe-link party host <party>`, one party at a time; the host loop ends when the party ends and the next poll
++// picks up the next one. Off unless party_pool.enabled is set in the Link's config file.
++func (n *Agent) pollPartyPool(ctx context.Context) {
++ if n.chatTokens == nil || n.opt.Creds == nil {
++ return
++ }
++ partyURL := n.opt.Config.PartyPool.PartyURL
++ if partyURL == "" {
++ partyURL = "https://openvibe.party"
++ }
++ client := party.Client{
++ PartyURL: partyURL,
++ ChatToken: func(ctx context.Context) (string, error) {
++ t, err := n.chatTokens.Token(ctx)
++ if err != nil {
++ return "", err
++ }
++ return t.Reveal(), nil
++ },
++ }
++ n.offerPartyPool(ctx, client)
++ tick := time.NewTicker(partyPoolInterval)
++ defer tick.Stop()
++ for {
++ select {
++ case <-ctx.Done():
++ return
++ case <-tick.C:
++ n.offerPartyPool(ctx, client)
++ }
++ }
++}
++
++// offerPartyPool starts the next assignment unless this Link is already hosting a party. A failure (Chat down, the
++// machine not registered, no container runtime) is logged and the poll tries again; it never affects the machine's
++// other work.
++func (n *Agent) offerPartyPool(ctx context.Context, client party.Client) {
++ n.mu.Lock()
++ busy := n.partyID != ""
++ n.mu.Unlock()
++ if busy || ctx.Err() != nil {
++ return
++ }
++ // The owner's stop is the owner's: no pool party starts while it is latched.
++ if st := n.latch.State(); st.Stopped() {
++ return
++ }
++ assignment, err := client.Assignment(ctx)
++ if err != nil {
++ n.log.Debug("party pool: no assignment", "err", err)
++ return
++ }
++ if assignment == nil || assignment.Party == "" {
++ return
++ }
++ if err := n.startParty(assignment.Party); err != nil {
++ n.log.Warn("party pool: the party could not start", "party", assignment.Party, "err", err)
++ return
++ }
++ n.log.Info("party pool: hosting", "party", assignment.Party)
++}
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex 645d2b1..8d04f1a 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -120,6 +120,16 @@ type Config struct {
LogLevel string `json:"log_level,omitempty"`
// Worker runs `function`, `code` and `linux` jobs from the control link (docs/worker.md). Off unless enabled.
Worker WorkerConfig `json:"worker,omitzero"`
++ // PartyPool offers this machine to OpenVibe's own computer pool for openvibe.party (docs/party-desktop.md).
++ // Off unless enabled. The machine still has to be registered by the owner on openvibe.party/pool.
++ PartyPool PartyPoolConfig `json:"party_pool,omitzero"`
++}
++
++// PartyPoolConfig is this machine's part in OpenVibe's own computer pool: the Link asks Chat, as this machine, what
++// the owner's pool is holding for it, and starts the same host path as `openvibe-link party host <party>`.
++type PartyPoolConfig struct {
++ Enabled bool `json:"enabled,omitempty"`
++ PartyURL string `json:"party_url,omitempty"` // default https://openvibe.party; loopback http for a test
}
// WorkerConfig is the job worker. Disabled (the default), every job is refused "class not available". Only the
@@@ -489,6 +499,13 @@ func (c *Config) Validate() error {
if c.Stream.LiveURL != "" && !strings.HasPrefix(c.Stream.LiveURL, "https://") {
return errors.New("config: stream.live_url must be an https:// origin")
}
++ if c.PartyPool.PartyURL != "" {
++ u, err := url.Parse(c.PartyPool.PartyURL)
++ loopback := err == nil && (u.Hostname() == "localhost" || u.Hostname() == "127.0.0.1" || u.Hostname() == "::1")
++ if err != nil || u.Host == "" || u.User != nil || (u.Scheme != "https" && !(u.Scheme == "http" && loopback)) {
++ return errors.New("config: party_pool.party_url must be an https:// origin (or loopback http for a test)")
++ }
++ }
for _, l := range []*float64{c.Limits.MaxSpeed, c.Limits.MaxTurn} {
if l != nil && (*l < 0 || *l > 1) {
return errors.New("config: limits must be between 0 and 1")
ddiff --git a/internal/config/config_test.go b/internal/config/config_test.go
iindex 0bb75f9..8dc4d09 100644
---- a/internal/config/config_test.go
++++ b/internal/config/config_test.go
@@@ -239,3 +239,29 @@ func TestWorkerDNS(t *testing.T) {
}
}
}
++
++func TestPartyPool(t *testing.T) {
++ p := filepath.Join(t.TempDir(), "config.json")
++ os.WriteFile(p, []byte(`{"party_pool":{"enabled":true}}`), 0o600)
++ c, err := Load(p)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if !c.PartyPool.Enabled || c.PartyPool.PartyURL != "" {
++ t.Fatalf("party_pool = %+v", c.PartyPool)
++ }
++ os.WriteFile(p, []byte(`{"party_pool":{"enabled":true,"party_url":"http://127.0.0.1:4400"}}`), 0o600)
++ if _, err := Load(p); err != nil {
++ t.Fatalf("loopback http refused: %v", err)
++ }
++ for _, body := range []string{
++ `{"party_pool":{"enabled":true,"party_url":"http://example.com"}}`,
++ `{"party_pool":{"enabled":true,"party_url":"not a url"}}`,
++ `{"party_pool":{"enabled":true,"party_url":"https://user:[email protected]"}}`,
++ } {
++ os.WriteFile(p, []byte(body), 0o600)
++ if _, err := Load(p); err == nil {
++ t.Errorf("accepted %s", body)
++ }
++ }
++}
ddiff --git a/internal/party/client.go b/internal/party/client.go
iindex feca185..51fff43 100644
---- a/internal/party/client.go
++++ b/internal/party/client.go
@@@ -170,3 +170,20 @@ func (c Client) SyncMembers(ctx context.Context, share Share, access Access) err
func (c Client) CloseShare(ctx context.Context, share Share) error {
return c.call(ctx, false, http.MethodDelete, "/api/v1/link/shares/"+url.PathEscape(share.ID)+"/party", nil, nil)
}
++
++// Assignment is one party the owner's pool is holding for this machine.
++type Assignment struct {
++ Party string `json:"party"`
++ Hold string `json:"hold"`
++ Title string `json:"title,omitempty"`
++}
++
++// Assignment asks Chat, as this machine, for the oldest party OpenVibe's pool is holding for it. A machine that the
++// owner did not register is answered null, never an error; the poll is also how the machine reports it is alive.
++func (c Client) Assignment(ctx context.Context) (*Assignment, error) {
++ var out struct {
++ Assignment *Assignment `json:"assignment"`
++ }
++ err := c.call(ctx, true, http.MethodGet, "/api/chat/parties/desktop/assignment", nil, &out)
++ return out.Assignment, err
++}
ddiff --git a/internal/party/client_test.go b/internal/party/client_test.go
nnew file mode 100644
iindex 0000000..a572971
---- /dev/null
++++ b/internal/party/client_test.go
@@@ -0,0 +1,45 @@
++package party
++
++import (
++ "context"
++ "io"
++ "net/http"
++ "net/http/httptest"
++ "testing"
++)
++
++func TestAssignmentAsksChatWithTheMachineToken(t *testing.T) {
++ var path, auth string
++ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++ path, auth = r.URL.Path, r.Header.Get("Authorization")
++ w.Header().Set("Content-Type", "application/json")
++ io.WriteString(w, `{"assignment":{"party":"party-0123abcd","hold":"ph_0123456789abcdef","title":"Friday film"}}`)
++ }))
++ defer srv.Close()
++ c := Client{PartyURL: srv.URL, ChatToken: func(context.Context) (string, error) { return "node-token", nil }}
++ got, err := c.Assignment(context.Background())
++ if err != nil {
++ t.Fatal(err)
++ }
++ if path != "/api/chat/parties/desktop/assignment" || auth != "Bearer node-token" {
++ t.Fatalf("asked %q with %q", path, auth)
++ }
++ if got == nil || got.Party != "party-0123abcd" || got.Hold != "ph_0123456789abcdef" || got.Title != "Friday film" {
++ t.Fatalf("assignment = %#v", got)
++ }
++}
++
++func TestAssignmentIsNilWhenThePoolHoldsNothing(t *testing.T) {
++ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
++ io.WriteString(w, `{"assignment":null}`)
++ }))
++ defer srv.Close()
++ c := Client{PartyURL: srv.URL, ChatToken: func(context.Context) (string, error) { return "node-token", nil }}
++ got, err := c.Assignment(context.Background())
++ if err != nil {
++ t.Fatal(err)
++ }
++ if got != nil {
++ t.Fatalf("assignment = %#v, want none", got)
++ }
++}