replication: implement websocket replication

js/docs/src/content/docs/lex-reference/broadcast/place-stream-broadcast-origin.md

···
 
 **Record Properties:**
 
-| Name          | Type     | Req'd | Description                                                                | Constraints        |
-| ------------- | -------- | ----- | -------------------------------------------------------------------------- | ------------------ |
-| `streamer`    | `string` | ✅    | DID of the streamer whose livestream is being published                    | Format: `did`      |
-| `server`      | `string` | ✅    | did of the server that's currently rebroadcasting the livestream           | Format: `did`      |
-| `broadcaster` | `string` | ❌    | did of the broadcaster that operates the server syndicating the livestream | Format: `did`      |
-| `updatedAt`   | `string` | ✅    | Periodically updated timestamp when this origin last saw a livestream      | Format: `datetime` |
-| `irohTicket`  | `string` | ❌    | Iroh ticket that can be used to access the livestream from the server      | Max Length: 2048   |
+| Name           | Type     | Req'd | Description                                                                | Constraints        |
+| -------------- | -------- | ----- | -------------------------------------------------------------------------- | ------------------ |
+| `streamer`     | `string` | ✅    | DID of the streamer whose livestream is being published                    | Format: `did`      |
+| `server`       | `string` | ✅    | did of the server that's currently rebroadcasting the livestream           | Format: `did`      |
+| `broadcaster`  | `string` | ❌    | did of the broadcaster that operates the server syndicating the livestream | Format: `did`      |
+| `updatedAt`    | `string` | ✅    | Periodically updated timestamp when this origin last saw a livestream      | Format: `datetime` |
+| `irohTicket`   | `string` | ❌    | Iroh ticket that can be used to access the livestream from the server      | Max Length: 2048   |
+| `websocketURL` | `string` | ❌    | URL of the websocket endpoint for the livestream                           | Format: `uri`      |
 
 ---
 
···
             "type": "string",
             "maxLength": 2048,
             "description": "Iroh ticket that can be used to access the livestream from the server"
+          },
+          "websocketURL": {
+            "type": "string",
+            "format": "uri",
+            "description": "URL of the websocket endpoint for the livestream"
           }
         }
       }

js/docs/src/content/docs/lex-reference/live/place-stream-live-subscribesegments.md

···
+---
+title: place.stream.live.subscribeSegments
+description: Reference for the place.stream.live.subscribeSegments lexicon
+---
+
+**Lexicon Version:** 1
+
+## Definitions
+
+<a name="main"></a>
+
+### `main`
+
+**Type:** `subscription`
+
+Subscribe to a stream's new segments as they come in!
+
+**Parameters:**
+
+| Name       | Type     | Req'd | Description                             | Constraints |
+| ---------- | -------- | ----- | --------------------------------------- | ----------- |
+| `streamer` | `string` | ✅    | The DID of the streamer to subscribe to |             |
+
+**Message:**
+
+- **Schema:**
+
+**Schema Type:** Union of:<br/>&nbsp;&nbsp;[`#segment`](#segment)
+
+---
+
+<a name="segment"></a>
+
+### `segment`
+
+**Type:** `bytes`
+
+MP4 file of a user's signed livestream segment
+
+---
+
+## Lexicon Source
+
+```json
+{
+  "lexicon": 1,
+  "id": "place.stream.live.subscribeSegments",
+  "defs": {
+    "main": {
+      "type": "subscription",
+      "description": "Subscribe to a stream's new segments as they come in!",
+      "parameters": {
+        "type": "params",
+        "required": ["streamer"],
+        "properties": {
+          "streamer": {
+            "type": "string",
+            "description": "The DID of the streamer to subscribe to"
+          }
+        }
+      },
+      "message": {
+        "schema": {
+          "type": "union",
+          "refs": ["#segment"]
+        }
+      },
+      "errors": []
+    },
+    "segment": {
+      "type": "bytes",
+      "description": "MP4 file of a user's signed livestream segment"
+    }
+  }
+}
+```

js/docs/src/content/docs/lex-reference/openapi.json

···
         ]
       }
     },
+    "/xrpc/place.stream.live.subscribeSegments": {
+      "get": {
+        "summary": "Subscribe to a stream's new segments as they come in!",
+        "operationId": "place.stream.live.subscribeSegments",
+        "tags": ["place.stream.live"],
+        "x-websocket": true,
+        "responses": {},
+        "parameters": [
+          {
+            "name": "streamer",
+            "in": "query",
+            "required": true,
+            "description": "The DID of the streamer to subscribe to",
+            "schema": {
+              "type": "string",
+              "description": "The DID of the streamer to subscribe to"
+            }
+          }
+        ],
+        "x-websocket-message": {
+          "schema": {
+            "oneOf": [
+              {
+                "$ref": "#/components/schemas/place.stream.live.subscribeSegments_segment"
+              }
+            ]
+          }
+        }
+      }
+    },
     "/xrpc/place.stream.graph.getFollowingUser": {
       "get": {
         "summary": "Get whether or not user A is following user B.",
···
           "record": {}
         },
         "required": ["cid", "record"]
+      },
+      "place.stream.live.subscribeSegments_segment": {
+        "type": "string",
+        "format": "byte",
+        "description": "MP4 file of a user's signed livestream segment"
       },
       "com.atproto.repo.strongRef": {
         "type": "object",

lexicons/place/stream/broadcast/origin.json

···
             "type": "string",
             "maxLength": 2048,
             "description": "Iroh ticket that can be used to access the livestream from the server"
+          },
+          "websocketURL": {
+            "type": "string",
+            "format": "uri",
+            "description": "URL of the websocket endpoint for the livestream"
           }
         }
       }

lexicons/place/stream/live/subscribeSegments.json

···
+{
+  "lexicon": 1,
+  "id": "place.stream.live.subscribeSegments",
+  "defs": {
+    "main": {
+      "type": "subscription",
+      "description": "Subscribe to a stream's new segments as they come in!",
+      "parameters": {
+        "type": "params",
+        "required": ["streamer"],
+        "properties": {
+          "streamer": {
+            "type": "string",
+            "description": "The DID of the streamer to subscribe to"
+          }
+        }
+      },
+      "message": {
+        "schema": {
+          "type": "union",
+          "refs": ["#segment"]
+        }
+      },
+      "errors": []
+    },
+    "segment": {
+      "type": "bytes",
+      "description": "MP4 file of a user's signed livestream segment"
+    }
+  }
+}

pkg/api/api.go

···
 		Recorder: metrics.NewRecorder(metrics.Config{}),
 	})
 	var xrpc http.Handler
-	xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.StatefulDB, a.op, mdlw, a.ATSync)
+	xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.StatefulDB, a.op, mdlw, a.ATSync, a.Bus)
 	if err != nil {
 		return nil, err
 	}

pkg/cmd/streamplace.go

···
 	"stream.place/streamplace/pkg/notifications"
 	"stream.place/streamplace/pkg/replication"
 	"stream.place/streamplace/pkg/replication/iroh_replicator"
+	"stream.place/streamplace/pkg/replication/websocketrep"
 	"stream.place/streamplace/pkg/rtmps"
 	v0 "stream.place/streamplace/pkg/schema/v0"
 	"stream.place/streamplace/pkg/spmetrics"
···
 		if err != nil {
 			return err
 		}
+	}
+	if slices.Contains(cli.Replicators, config.ReplicatorWebsocket) {
+		replicator = websocketrep.NewWebsocketReplicator(b, mod, mm)
 	}
 
 	op := oatproxy.New(&oatproxy.Config{

pkg/config/config.go

···
 	DevAccountCreds            map[string]string
 	StreamSessionTimeout       time.Duration
 	Replicators                []string
+	WebsocketURL               string
 }
 
 // ContentFilters represents the content filtering configuration
···
 }
 
 const (
-	ReplicatorHTTP string = "http"
-	ReplicatorIroh string = "iroh"
+	ReplicatorWebsocket string = "websocket"
+	ReplicatorIroh      string = "iroh"
 )
 
 func (cli *CLI) NewFlagSet(name string) *flag.FlagSet {
···
 	fs.BoolVar(&cli.DisableIrohRelay, "disable-iroh-relay", false, "disable the iroh relay")
 	cli.KVSliceFlag(fs, &cli.DevAccountCreds, "dev-account-creds", "", "(FOR DEVELOPMENT ONLY) did=password pairs for logging into test accounts without oauth")
 	fs.DurationVar(&cli.StreamSessionTimeout, "stream-session-timeout", 60*time.Second, "how long to wait before considering a stream inactive on this node?")
-	cli.StringSliceFlag(fs, &cli.Replicators, "replicators", []string{ReplicatorIroh}, "list of replication protocols to use (http, iroh)")
+	cli.StringSliceFlag(fs, &cli.Replicators, "replicators", []string{ReplicatorWebsocket}, "list of replication protocols to use (http, iroh)")
+	fs.StringVar(&cli.WebsocketURL, "websocket-url", "", "override the websocket (ws:// or wss://) url to use for replication (normally not necessary, used for testing)")
 
 	lpFlags := flag.NewFlagSet("livepeer", flag.ContinueOnError)
 	_ = starter.NewLivepeerConfig(lpFlags)

pkg/media/validate.go

···
 	if err != nil {
 		return err
 	}
+	oldSeg, err := mm.model.GetSegment(*maniCert.Manifest.Label)
+	if err != nil {
+		return err
+	}
+	if oldSeg != nil {
+		log.Warn(ctx, "segment already exists, skipping", "segmentID", *maniCert.Manifest.Label)
+		return nil
+	}
 	pub, err := signers.ParseES256KCert([]byte(maniCert.Cert))
 	if err != nil {
 		return err

pkg/multitest/multitest_test.go

···
 	for _, acct := range dev.Accounts {
 		devAccountCreds = append(devAccountCreds, fmt.Sprintf("%s=%s", acct.DID, acct.Password))
 	}
+	apiPort := nextPort()
 	env := map[string]string{
-		"SP_HTTP_ADDR":              fmt.Sprintf("127.0.0.1:%d", nextPort()),
+		"SP_HTTP_ADDR":              fmt.Sprintf("127.0.0.1:%d", apiPort),
 		"SP_HTTP_INTERNAL_ADDR":     fmt.Sprintf("127.0.0.1:%d", nextPort()),
 		"SP_RELAY_HOST":             strings.ReplaceAll(dev.PDSURL, "http://", "ws://"),
 		"SP_PLC_URL":                dev.PLCURL,
···
 		"SP_STREAM_SESSION_TIMEOUT": "3s",
 		"SP_COLOR":                  "true",
 		"RUST_LOG":                  os.Getenv("RUST_LOG"),
+		"SP_BROADCASTER_HOST":       fmt.Sprintf("%s.example.com", name),
+		"SP_WEBSOCKET_URL":          fmt.Sprintf("ws://127.0.0.1:%d", apiPort),
 	}
 	_, file, _, _ := runtime.Caller(0)
 	buildDir := fmt.Sprintf("build-%s-%s", runtime.GOOS, runtime.GOARCH)

pkg/replication/websocketrep/websocket_replicator.go

···
+package websocketrep
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/url"
+	"sync"
+
+	"github.com/gorilla/websocket"
+	"golang.org/x/sync/errgroup"
+	"stream.place/streamplace/pkg/bus"
+	"stream.place/streamplace/pkg/config"
+	"stream.place/streamplace/pkg/log"
+	"stream.place/streamplace/pkg/media"
+	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/streamplace"
+)
+
+type WebsocketReplicator struct {
+	bus        *bus.Bus
+	cli        *config.CLI
+	mod        model.Model
+	conns      map[string]bool
+	connsMutex sync.RWMutex
+	group      *errgroup.Group
+	mm         *media.MediaManager
+}
+
+func NewWebsocketReplicator(bus *bus.Bus, mod model.Model, mm *media.MediaManager) *WebsocketReplicator {
+	return &WebsocketReplicator{
+		bus:        bus,
+		mod:        mod,
+		conns:      make(map[string]bool),
+		connsMutex: sync.RWMutex{},
+		mm:         mm,
+	}
+}
+
+func (r *WebsocketReplicator) Start(ctx context.Context, cli *config.CLI) error {
+	r.cli = cli
+	_ = r.getMyWebsocketURL() // panic check
+	r.group, ctx = errgroup.WithContext(ctx)
+	return r.startBusSubscribe(ctx)
+}
+
+func (r *WebsocketReplicator) startBusSubscribe(ctx context.Context) error {
+	// start subscription first so we're buffering new origins
+	busCh := r.bus.Subscribe("")
+	originViews, err := r.mod.GetRecentBroadcastOrigins(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get recent broadcast origins: %w", err)
+	}
+	for _, view := range originViews {
+		err = r.handleOriginMessage(ctx, view)
+		if err != nil {
+			log.Error(ctx, "could not check origin", "error", err)
+		}
+	}
+	log.Log(ctx, "Resumed recent broadcast origins", "count", len(originViews))
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case msg := <-busCh:
+			if view, ok := msg.(*streamplace.BroadcastDefs_BroadcastOriginView); ok {
+				log.Debug(ctx, "got broadcast origin view", "view", view)
+				err = r.handleOriginMessage(ctx, view)
+				if err != nil {
+					log.Error(ctx, "could not handle origin message", "error", err)
+				}
+			}
+		}
+	}
+}
+
+func (r *WebsocketReplicator) handleOriginMessage(ctx context.Context, view *streamplace.BroadcastDefs_BroadcastOriginView) error {
+	origin, ok := view.Record.Val.(*streamplace.BroadcastOrigin)
+	if !ok {
+		return fmt.Errorf("record is not a BroadcastOrigin")
+	}
+	ctx = log.WithLogValues(ctx, "streamer", view.Author.Did)
+	if origin.WebsocketURL == nil {
+		return fmt.Errorf("origin has no websocket URL author=%s", view.Author.Did)
+	}
+	if r.hasConnection(origin.Streamer) {
+		log.Debug(ctx, "already has connection")
+		return nil
+	}
+	myURL := r.getMyWebsocketURL()
+	u, err := url.Parse(*origin.WebsocketURL)
+	if err != nil {
+		return fmt.Errorf("could not parse origin websocket URL: %w", err)
+	}
+	if u.Host == myURL.Host {
+		log.Debug(ctx, "origin websocket URL is on this node, skipping")
+		return nil
+	}
+	r.group.Go(func() error {
+		err := r.openWebsocket(ctx, view)
+		log.Error(ctx, "websocket connection error", "error", err)
+		return nil
+	})
+	return nil
+}
+
+func (r *WebsocketReplicator) openWebsocket(ctx context.Context, view *streamplace.BroadcastDefs_BroadcastOriginView) error {
+	err := r.tryConnection(view.Author.Did)
+	if err != nil {
+		return err
+	}
+	defer r.removeConnection(view.Author.Did)
+	origin, ok := view.Record.Val.(*streamplace.BroadcastOrigin)
+	if !ok {
+		return fmt.Errorf("record is not a BroadcastOrigin")
+	}
+	if origin.WebsocketURL == nil {
+		return fmt.Errorf("origin has no websocket URL")
+	}
+	conn, _, err := websocket.DefaultDialer.Dial(*origin.WebsocketURL, nil)
+	if err != nil {
+		return fmt.Errorf("could not dial websocket: %w", err)
+	}
+	defer conn.Close()
+	for {
+		typ, msg, err := conn.ReadMessage()
+		if err != nil {
+			log.Error(ctx, "could not read message", "error", err)
+			return fmt.Errorf("could not read message: %w", err)
+		}
+		if typ != websocket.BinaryMessage {
+			log.Error(ctx, "expected binary message", "type", typ)
+			return fmt.Errorf("expected binary message")
+		}
+		log.Debug(ctx, "received message", "type", typ, "length", len(msg))
+		err = r.mm.ValidateMP4(context.Background(), bytes.NewReader(msg), false)
+		if err != nil {
+			return fmt.Errorf("could not validate segment: %w", err)
+		}
+	}
+}
+
+func (r *WebsocketReplicator) hasConnection(origin string) bool {
+	r.connsMutex.RLock()
+	defer r.connsMutex.RUnlock()
+	return r.conns[origin]
+}
+
+func (r *WebsocketReplicator) tryConnection(origin string) error {
+	r.connsMutex.Lock()
+	defer r.connsMutex.Unlock()
+	if _, ok := r.conns[origin]; ok {
+		return fmt.Errorf("connection already exists")
+	}
+	r.conns[origin] = true
+	return nil
+}
+
+func (r *WebsocketReplicator) removeConnection(origin string) {
+	r.connsMutex.Lock()
+	defer r.connsMutex.Unlock()
+	delete(r.conns, origin)
+}
+
+// we're pull-based, nothing to do here
+func (r *WebsocketReplicator) SendSegment(ctx context.Context, seg *media.NewSegmentNotification) error {
+	return nil
+}
+
+func (r *WebsocketReplicator) BuildOriginRecord(origin *streamplace.BroadcastOrigin) error {
+	u := r.getMyWebsocketURL()
+	u.Path = "/xrpc/place.stream.live.subscribeSegments"
+	u.RawQuery = url.Values{
+		"streamer": []string{origin.Streamer},
+	}.Encode()
+
+	urlStr := u.String()
+	origin.WebsocketURL = &urlStr
+	return nil
+}
+
+func (r *WebsocketReplicator) getMyWebsocketURL() *url.URL {
+	if r.cli.WebsocketURL != "" {
+		u, err := url.Parse(r.cli.WebsocketURL)
+		// chill to panic, we're going to check this on boot
+		if err != nil {
+			panic("invalid websocket override URL: " + r.cli.WebsocketURL)
+		}
+		return u
+	}
+	u := url.URL{
+		Scheme: "ws",
+		Host:   r.cli.ServerHost,
+	}
+	if r.cli.Secure {
+		u.Scheme = "wss"
+	}
+	return &u
+}

pkg/spmetrics/spmetrics.go

···
 
 var WebsocketsOpen = promauto.NewGauge(prometheus.GaugeOpts{
 	Name: "streamplace_websockets_open",
-	Help: "number of open websockets",
+	Help: "number of open playback websockets",
+})
+
+var ReplicationWebsocketsOpen = promauto.NewGauge(prometheus.GaugeOpts{
+	Name: "streamplace_replication_websockets_open",
+	Help: "number of open replication websockets",
 })
 
 var SegmentSubscriptionsOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{

pkg/spxrpc/place_stream_live.go

···
 	"time"
 
 	"github.com/bluesky-social/indigo/lex/util"
+	"github.com/gorilla/websocket"
 	"github.com/labstack/echo/v4"
+	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/spid"
 	"stream.place/streamplace/pkg/spmetrics"
 
 	placestreamtypes "stream.place/streamplace/pkg/streamplace"
 )
+
+var replicationUpgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024 * 1024 * 10, // 10MB
+	CheckOrigin: func(r *http.Request) bool {
+		return true
+	},
+}
 
 func (s *Server) handlePlaceStreamLiveGetSegments(ctx context.Context, before string, limit int, userDID string) (*placestreamtypes.LiveGetSegments_Output, error) {
 	if userDID == "" {
···
 
 	return liveUsers, nil
 }
+
+func (s *Server) handlePlaceStreamLiveSubscribeSegments(c echo.Context) error {
+	user := c.QueryParam("streamer")
+	if user == "" {
+		return echo.NewHTTPError(http.StatusBadRequest, "User DID is required")
+	}
+	spmetrics.ReplicationWebsocketsOpen.Inc()
+	defer spmetrics.ReplicationWebsocketsOpen.Dec()
+	ws, err := replicationUpgrader.Upgrade(c.Response(), c.Request(), nil)
+	if err != nil {
+		return err
+	}
+	defer ws.Close()
+
+	ctx, cancel := context.WithCancel(c.Request().Context())
+	defer cancel()
+	go func() {
+
+		segChan := s.bus.SubscribeSegmentBuf(ctx, user, "source", 2)
+		defer s.bus.UnsubscribeSegment(ctx, user, "source", segChan)
+		for {
+			select {
+			case <-ctx.Done():
+				log.Debug(ctx, "exiting segment reader")
+				return
+			case file := <-segChan.C:
+				log.Debug(ctx, "got segment", "file", file.Filepath)
+				err := ws.WriteMessage(websocket.BinaryMessage, file.Data)
+				if err != nil {
+					log.Error(ctx, "could not write message", "error", err)
+					cancel()
+					return
+				}
+			}
+		}
+
+	}()
+
+	for {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		// Read
+		_, msg, err := ws.ReadMessage()
+		if err != nil {
+			c.Logger().Error(err)
+			return err
+		}
+		log.Debug(c.Request().Context(), "received message", "message", string(msg))
+	}
+}

pkg/spxrpc/spxrpc.go

···
 	"github.com/streamplace/oatproxy/pkg/oatproxy"
 	"stream.place/streamplace/pkg/aqhttp"
 	"stream.place/streamplace/pkg/atproto"
+	"stream.place/streamplace/pkg/bus"
 	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/model"
···
 	OGImageCache *cache.Cache
 	ATSync       *atproto.ATProtoSynchronizer
 	statefulDB   *statedb.StatefulDB
+	bus          *bus.Bus
 }
 
-func NewServer(ctx context.Context, cli *config.CLI, model model.Model, statefulDB *statedb.StatefulDB, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer) (*Server, error) {
+func NewServer(ctx context.Context, cli *config.CLI, model model.Model, statefulDB *statedb.StatefulDB, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer, bus *bus.Bus) (*Server, error) {
 	e := echo.New()
 	s := &Server{
 		e:            e,
···
 		OGImageCache: cache.New(5*time.Minute, 10*time.Minute), // 5min TTL, 10min cleanup
 		ATSync:       atsync,
 		statefulDB:   statefulDB,
+		bus:          bus,
 	}
 	e.Use(s.ErrorHandlingMiddleware())
 	e.Use(s.ContextPreservingMiddleware())
···
 		return c.JSON(http.StatusOK, map[string]string{"version": cli.Build.Version})
 	})
 	e.GET("/xrpc/com.atproto.sync.subscribeRepos", s.handleComAtprotoSyncSubscribeRepos)
+	e.GET("/xrpc/place.stream.live.subscribeSegments", s.handlePlaceStreamLiveSubscribeSegments)
 	e.GET("/xrpc/*", s.HandleWildcard)
 	e.POST("/xrpc/*", s.HandleWildcard)
 	return s, nil

pkg/streamplace/broadcastorigin.go

···
 	Streamer string `json:"streamer" cborgen:"streamer"`
 	// updatedAt: Periodically updated timestamp when this origin last saw a livestream
 	UpdatedAt string `json:"updatedAt" cborgen:"updatedAt"`
+	// websocketURL: URL of the websocket endpoint for the livestream
+	WebsocketURL *string `json:"websocketURL,omitempty" cborgen:"websocketURL,omitempty"`
 }

pkg/streamplace/cbor_gen.go

···
 	}
 
 	cw := cbg.NewCborWriter(w)
-	fieldCount := 6
+	fieldCount := 7
 
 	if t.Broadcaster == nil {
 		fieldCount--
 	}
 
 	if t.IrohTicket == nil {
+		fieldCount--
+	}
+
+	if t.WebsocketURL == nil {
 		fieldCount--
 	}
 
···
 			}
 		}
 	}
+
+	// t.WebsocketURL (string) (string)
+	if t.WebsocketURL != nil {
+
+		if len("websocketURL") > 1000000 {
+			return xerrors.Errorf("Value in field \"websocketURL\" was too long")
+		}
+
+		if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("websocketURL"))); err != nil {
+			return err
+		}
+		if _, err := cw.WriteString(string("websocketURL")); err != nil {
+			return err
+		}
+
+		if t.WebsocketURL == nil {
+			if _, err := cw.Write(cbg.CborNull); err != nil {
+				return err
+			}
+		} else {
+			if len(*t.WebsocketURL) > 1000000 {
+				return xerrors.Errorf("Value in field t.WebsocketURL was too long")
+			}
+
+			if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(*t.WebsocketURL))); err != nil {
+				return err
+			}
+			if _, err := cw.WriteString(string(*t.WebsocketURL)); err != nil {
+				return err
+			}
+		}
+	}
 	return nil
 }
 
···
 
 	n := extra
 
-	nameBuf := make([]byte, 11)
+	nameBuf := make([]byte, 12)
 	for i := uint64(0); i < n; i++ {
 		nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 1000000)
 		if err != nil {
···
 					}
 
 					t.Broadcaster = (*string)(&sval)
+				}
+			}
+			// t.WebsocketURL (string) (string)
+		case "websocketURL":
+
+			{
+				b, err := cr.ReadByte()
+				if err != nil {
+					return err
+				}
+				if b != cbg.CborNull[0] {
+					if err := cr.UnreadByte(); err != nil {
+						return err
+					}
+
+					sval, err := cbg.ReadStringWithMax(cr, 1000000)
+					if err != nil {
+						return err
+					}
+
+					t.WebsocketURL = (*string)(&sval)
 				}
 			}
 

pkg/streamplace/livesubscribeSegments.go

···
+// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT.
+
+package streamplace
+
+// schema: place.stream.live.subscribeSegments