From 4cf9880d875e8bdd041525fcffb2c55cfac64d4a Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Mon, 23 Feb 2026 16:28:18 -0800 Subject: [PATCH] state: add broadcastorigin record --- pkg/atproto/lexicon_repo.go | 6 +-- pkg/atproto/lexicon_repo_test.go | 10 ++-- pkg/config/config.go | 6 ++- pkg/director/stream_session.go | 4 ++ pkg/media/manifest_builder.go | 9 ++++ .../websocketrep/websocket_replicator.go | 2 +- pkg/spxrpc/com_atproto_repo.go | 4 +- pkg/spxrpc/spxrpc.go | 4 +- pkg/statedb/broadcast_origin.go | 51 +++++++++++++++++++ pkg/statedb/statedb.go | 1 + 10 files changed, 83 insertions(+), 14 deletions(-) create mode 100644 pkg/statedb/broadcast_origin.go diff --git a/pkg/atproto/lexicon_repo.go b/pkg/atproto/lexicon_repo.go index 9d2370fb..43c611fe 100644 --- a/pkg/atproto/lexicon_repo.go +++ b/pkg/atproto/lexicon_repo.go @@ -195,7 +195,7 @@ func MakeLexiconRepo(ctx context.Context, cli *config.CLI, mod model.Model, stat return priv.HashAndSign(sb) } - events, err := state.GetCommitEventsSince(cli.MyDID(), time.Time{}) + events, err := state.GetCommitEventsSince(cli.BroadcasterDID(), time.Time{}) if err != nil { return nil, fmt.Errorf("failed to get commit events: %w", err) } @@ -222,7 +222,7 @@ func MakeLexiconRepo(ctx context.Context, cli *config.CLI, mod model.Model, stat if err != nil { return nil, fmt.Errorf("failed to create delta session: %w", err) } - LexiconRepo = atrepo.NewRepo(ctx, cli.MyDID(), ses) + LexiconRepo = atrepo.NewRepo(ctx, cli.BroadcasterDID(), ses) } else { LexiconRepo, err = atrepo.OpenRepo(ctx, ses, currentRoot) if err != nil { @@ -316,7 +316,7 @@ func MakeLexiconRepo(ctx context.Context, cli *config.CLI, mod model.Model, stat if len(ops) > 0 { log.Log(ctx, "created new lexicon commit for changes", "did", signed.Did, "data", signed.Data, "prev", signed.Prev, "rev", signed.Rev) commit := &comatproto.SyncSubscribeRepos_Commit{ - Repo: cli.MyDID(), + Repo: cli.BroadcasterDID(), Blocks: blocks, Rev: currentRev, Commit: lexutil.LexLink(currentRoot), diff --git a/pkg/atproto/lexicon_repo_test.go b/pkg/atproto/lexicon_repo_test.go index f014701a..14c611e1 100644 --- a/pkg/atproto/lexicon_repo_test.go +++ b/pkg/atproto/lexicon_repo_test.go @@ -40,10 +40,10 @@ func TestLexiconRepo(t *testing.T) { require.NotNil(t, rec) handle.Close() - evts, err := state.GetCommitEventsSinceSeq(cli.MyDID(), 0) + evts, err := state.GetCommitEventsSinceSeq(cli.BroadcasterDID(), 0) require.NoError(t, err) require.Len(t, evts, 1) - require.Equal(t, evts[0].RepoDID, cli.MyDID()) + require.Equal(t, evts[0].RepoDID, cli.BroadcasterDID()) // opening an existing repo handle, err = MakeLexiconRepo(context.Background(), &cli, mod, state) @@ -100,11 +100,11 @@ func TestLexiconRepo(t *testing.T) { require.NoError(t, err) handle.Close() - evts, err = state.GetCommitEventsSinceSeq(cli.MyDID(), 0) + evts, err = state.GetCommitEventsSinceSeq(cli.BroadcasterDID(), 0) require.NoError(t, err) require.Len(t, evts, 2) - require.Equal(t, evts[0].RepoDID, cli.MyDID()) - require.Equal(t, evts[1].RepoDID, cli.MyDID()) + require.Equal(t, evts[0].RepoDID, cli.BroadcasterDID()) + require.Equal(t, evts[1].RepoDID, cli.BroadcasterDID()) oldCommit, err := evts[0].ToCommitEvent() require.NoError(t, err) newCommit, err := evts[1].ToCommitEvent() diff --git a/pkg/config/config.go b/pkg/config/config.go index b9e268be..66e9fbe2 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -1152,10 +1152,14 @@ func (cli *CLI) StreamIsAllowed(did string) error { return fmt.Errorf("user is not allowed to stream") } -func (cli *CLI) MyDID() string { +func (cli *CLI) BroadcasterDID() string { return fmt.Sprintf("did:web:%s", cli.BroadcasterHost) } +func (cli *CLI) ServerDID() string { + return fmt.Sprintf("did:web:%s", cli.ServerHost) +} + func (cli *CLI) HasHTTPS() bool { return cli.Secure || cli.BehindHTTPSProxy } diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index d16abdae..324da01c 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -214,6 +214,10 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment }) }) + ss.Go(ctx, func() error { + return ss.statefulDB.UpsertBroadcastOrigin(spseg.Creator, ss.cli.BroadcasterDID(), time.Now()) + }) + // everything else is for published segments if !notif.Metadata.Published { return nil diff --git a/pkg/media/manifest_builder.go b/pkg/media/manifest_builder.go index c3902b6f..1740798c 100644 --- a/pkg/media/manifest_builder.go +++ b/pkg/media/manifest_builder.go @@ -156,6 +156,15 @@ func (mb *ManifestBuilder) BuildManifest(ctx context.Context, streamerName strin // Update the manifest title with the retrieved livestream title mani["assertions"].([]obj)[1]["data"].(obj)["dc:title"] = livestreamTitle + if ls != nil { + mani["assertions"] = append(mani["assertions"].([]obj), obj{ + "label": "place.stream.livestream", + "data": ls, + }) + } else { + log.Warn(ctx, "ManifestBuilder: no livestream found for streamer", "did", streamerName) + } + // Convert manifest to JSON bytes for use with Rust c2pa library manifestBs, err := json.Marshal(mani) if err != nil { diff --git a/pkg/replication/websocketrep/websocket_replicator.go b/pkg/replication/websocketrep/websocket_replicator.go index 0a6b1fd5..e51e0ca2 100644 --- a/pkg/replication/websocketrep/websocket_replicator.go +++ b/pkg/replication/websocketrep/websocket_replicator.go @@ -123,7 +123,7 @@ func (r *WebsocketReplicator) openWebsocket(ctx context.Context, view *streampla } conn, _, err := websocket.DefaultDialer.Dial(*origin.WebsocketURL, nil) if err != nil { - return fmt.Errorf("could not dial websocket: %w", err) + return fmt.Errorf("could not dial websocket (%s): %w", *origin.WebsocketURL, err) } defer conn.Close() for { diff --git a/pkg/spxrpc/com_atproto_repo.go b/pkg/spxrpc/com_atproto_repo.go index 328376d3..52d41038 100644 --- a/pkg/spxrpc/com_atproto_repo.go +++ b/pkg/spxrpc/com_atproto_repo.go @@ -92,8 +92,8 @@ func (s *Server) handleComAtprotoRepoDescribeRepo(ctx context.Context, repo stri } return &comatproto.RepoDescribeRepo_Output{ - Handle: s.cli.MyDID(), - Did: s.cli.MyDID(), + Handle: s.cli.BroadcasterDID(), + Did: s.cli.BroadcasterDID(), DidDoc: atproto.DIDDoc(s.cli.BroadcasterHost), Collections: []string{ "com.atproto.lexicon.schema", diff --git a/pkg/spxrpc/spxrpc.go b/pkg/spxrpc/spxrpc.go index cf00df19..8265630d 100644 --- a/pkg/spxrpc/spxrpc.go +++ b/pkg/spxrpc/spxrpc.go @@ -85,7 +85,7 @@ func (s *Server) isLocalPDS(ctx context.Context, repo string) (bool, string, err if err != nil { return false, "", fmt.Errorf("resolveRepoService: %w", err) } - if did == s.cli.MyDID() { + if did == s.cli.BroadcasterDID() { return true, svc, nil } return false, svc, nil @@ -106,7 +106,7 @@ func makeUnauthenticatedRequest(ctx context.Context, service, method string, par } u.RawQuery = query.Encode() - log.Error(ctx, "making unauthenticated request", "url", u.String()) + log.Debug(ctx, "making unauthenticated request", "url", u.String()) req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil) if err != nil { diff --git a/pkg/statedb/broadcast_origin.go b/pkg/statedb/broadcast_origin.go new file mode 100644 index 00000000..c53691a9 --- /dev/null +++ b/pkg/statedb/broadcast_origin.go @@ -0,0 +1,51 @@ +package statedb + +import ( + "time" +) + +type BroadcastOrigin struct { + StreamerRepoDID string `gorm:"column:streamer_repo_did;primarykey;index:idx_streamer_repo_did_updated_at,priority:1"` + ServerDID string `gorm:"column:server_did;primarykey;index:idx_server_did_updated_at,priority:1"` + UpdatedAt time.Time `gorm:"column:updated_at;index:idx_streamer_repo_did_updated_at,priority:2;index:idx_server_did_updated_at,priority:2"` +} + +func (m *BroadcastOrigin) TableName() string { + return "broadcast_origins" +} + +// UpsertBroadcastOrigin inserts or updates a BroadcastOrigin entry. +// If an entry with the same StreamerRepoDID and ServerRepoDID exists, it updates UpdatedAt. +// Otherwise, it creates a new entry. +func (state *StatefulDB) UpsertBroadcastOrigin(streamerRepoDID, serverRepoDID string, updatedAt time.Time) error { + broadcastOrigin := &BroadcastOrigin{ + StreamerRepoDID: streamerRepoDID, + ServerDID: serverRepoDID, + UpdatedAt: updatedAt, + } + // Uses GORM's upsert ("ON CONFLICT DO UPDATE") by providing primary keys and using Updates + return state.DB. + Clauses( + // GORM uses these settings to upsert + // The clause 'ON CONFLICT (primary key) DO UPDATE' is default when calling Save + ). + Save(broadcastOrigin).Error +} + +// GetLatestBroadcastOriginForStreamer retrieves the most recent BroadcastOrigin for a given streamerRepoDID, +// ordered by UpdatedAt descending, and returns the first found. +func (state *StatefulDB) GetLatestBroadcastOriginForStreamer(streamerRepoDID string) (*BroadcastOrigin, error) { + var origin BroadcastOrigin + tx := state.DB. + Where("streamer_repo_did = ?", streamerRepoDID). + Order("updated_at DESC"). + Limit(1). + Find(&origin) + if tx.Error != nil { + return nil, tx.Error + } + if tx.RowsAffected == 0 { + return nil, nil + } + return &origin, nil +} diff --git a/pkg/statedb/statedb.go b/pkg/statedb/statedb.go index c1e06e1f..604fbf08 100644 --- a/pkg/statedb/statedb.go +++ b/pkg/statedb/statedb.go @@ -53,6 +53,7 @@ var StatefulDBModels = []any{ MultistreamEvent{}, BrandingBlob{}, ModerationAuditLog{}, + BroadcastOrigin{}, } var NoPostgresDatabaseCode = "3D000" -- 2.51.2