diff --git a/pkg/atproto/profile.go b/pkg/atproto/profile.go new file mode 100644 index 000000000..cdfd77396 --- /dev/null +++ b/pkg/atproto/profile.go @@ -0,0 +1,45 @@ +package atproto + +import ( + "context" + "errors" + "fmt" + + "github.com/bluesky-social/indigo/api/bsky" + "github.com/bluesky-social/indigo/xrpc" +) + +var ErrUserNotFound = errors.New("user not found") + +func (atsync *ATProtoSynchronizer) FetchUserProfile(ctx context.Context, username string) (*bsky.ActorDefs_ProfileViewDetailed, error) { + // Use ATSync to resolve username to DID, then fetch full profile from Bluesky + var actor string + + // First try to resolve via internal DB + repo, err := atsync.Model.GetRepoByHandleOrDID(username) + if err != nil { + return nil, fmt.Errorf("%w: %w", ErrUserNotFound, err) + } else if repo != nil { + // Use the DID as it's the most reliable identifier + actor = repo.DID + } else { + return nil, fmt.Errorf("no repo found for username: %s (%w)", username, ErrUserNotFound) + } + + // Fetch full profile from Bluesky public API + client := &xrpc.Client{ + Host: "https://public.api.bsky.app", + } + + profile, err := bsky.ActorGetProfile(ctx, client, actor) + if err != nil { + return nil, fmt.Errorf("failed to fetch profile from Bluesky for '%s': %w", actor, err) + } + + if profile == nil { + return nil, fmt.Errorf("received nil profile from Bluesky API for '%s'", actor) + } + + return profile, nil + +} diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index c82df90c5..74c6fb781 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -318,7 +318,7 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu Public: cli.PublicOAuth, HTTPClient: &aqhttp.Client, }) - d := director.NewDirector(mm, mod, cli, b, op, state, replicator, ldb) + d := director.NewDirector(mm, mod, cli, b, op, state, replicator, ldb, atsync) a, err := api.MakeStreamplaceAPI(cli, mod, state, noter, mm, ms, b, atsync, d, op, ldb) if err != nil { return err diff --git a/pkg/director/director.go b/pkg/director/director.go index c12d22bf5..129c5599d 100644 --- a/pkg/director/director.go +++ b/pkg/director/director.go @@ -7,6 +7,7 @@ import ( "github.com/streamplace/oatproxy/pkg/oatproxy" "golang.org/x/sync/errgroup" + "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/localdb" @@ -34,9 +35,10 @@ type Director struct { statefulDB *statedb.StatefulDB replicator replication.Replicator localDB localdb.LocalDB + atsync *atproto.ATProtoSynchronizer } -func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy, statefulDB *statedb.StatefulDB, replicator replication.Replicator, ldb localdb.LocalDB) *Director { +func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy, statefulDB *statedb.StatefulDB, replicator replication.Replicator, ldb localdb.LocalDB, atsync *atproto.ATProtoSynchronizer) *Director { return &Director{ mm: mm, mod: mod, @@ -48,6 +50,7 @@ func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus * statefulDB: statefulDB, replicator: replicator, localDB: ldb, + atsync: atsync, } } @@ -83,6 +86,7 @@ func (d *Director) Start(ctx context.Context) error { statusUpdateChan: make(chan struct{}, 1), originUpdateChan: make(chan struct{}, 1), localDB: d.localDB, + atsync: d.atsync, } d.streamSessions[not.Segment.RepoDID] = ss g.Go(func() error { diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index d34ba0991..37d5faf9c 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -17,6 +17,7 @@ import ( "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/aqhttp" "stream.place/streamplace/pkg/aqtime" + "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/livepeer" @@ -57,6 +58,7 @@ type StreamSession struct { packets []bus.PacketizedSegment statefulDB *statedb.StatefulDB replicator replication.Replicator + atsync *atproto.ATProtoSynchronizer } func (ss *StreamSession) Start(ctx context.Context, notif *media.NewSegmentNotification) error { @@ -129,6 +131,15 @@ func (ss *StreamSession) Start(ctx context.Context, notif *media.NewSegmentNotif }) } + ss.Go(ctx, func() error { + err := ss.UpdateProfilePicture(ctx, spseg.Creator) + if err != nil { + log.Error(ctx, "failed to update profile picture", "error", err) + } + <-ctx.Done() + return nil + }) + for { select { case <-ss.segmentChan: diff --git a/pkg/director/stream_session_redcircle.go b/pkg/director/stream_session_redcircle.go new file mode 100644 index 000000000..5b50d653c --- /dev/null +++ b/pkg/director/stream_session_redcircle.go @@ -0,0 +1,120 @@ +package director + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/api/bsky" + lexutil "github.com/bluesky-social/indigo/lex/util" + + "github.com/bluesky-social/indigo/xrpc" + "stream.place/streamplace/pkg/aqhttp" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/redcircle" +) + +func downloadImage(ctx context.Context, url string) ([]byte, error) { + if url == "" { + return nil, errors.New("empty URL provided") + } + + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + resp, err := aqhttp.Do(ctx, req) + if err != nil { + return nil, fmt.Errorf("HTTP request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("HTTP request failed with status %d: %s", resp.StatusCode, resp.Status) + } + + imageData, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("failed to read image data: %w", err) + } + + return imageData, nil +} + +func (ss *StreamSession) UpdateProfilePicture(ctx context.Context, repoDID string) error { + session, err := ss.statefulDB.GetSessionByDID(ss.repoDID) + if err != nil { + return fmt.Errorf("failed to get session: %w", err) + } + if session == nil { + return fmt.Errorf("no session found for repoDID: %s", ss.repoDID) + } + session, err = ss.op.RefreshIfNeeded(session) + if err != nil { + return fmt.Errorf("failed to refresh session: %w", err) + } + client, err := ss.op.GetXrpcClient(session) + if err != nil { + return fmt.Errorf("failed to get xrpc client: %w", err) + } + profile, err := ss.atsync.FetchUserProfile(ctx, repoDID) + if err != nil { + return fmt.Errorf("failed to fetch user profile: %w", err) + } + if profile == nil { + return fmt.Errorf("received nil profile from Bluesky API for '%s'", repoDID) + } + if profile.Avatar == nil || *profile.Avatar == "" { + return fmt.Errorf("received nil or empty avatar from Bluesky API for '%s'", repoDID) + } + profileAvatar := *profile.Avatar + imageData, err := downloadImage(ctx, profileAvatar) + if err != nil { + return fmt.Errorf("failed to download image: %w", err) + } + circleImage, err := redcircle.GenerateRedCircle(ctx, imageData) + if err != nil { + return fmt.Errorf("failed to generate redcircle: %w", err) + } + var out comatproto.RepoUploadBlob_Output + err = client.Do(ctx, xrpc.Procedure, "image/jpeg", "com.atproto.repo.uploadBlob", nil, bytes.NewReader(circleImage), &out) + if err != nil { + return fmt.Errorf("failed to upload blob: %w", err) + } + if out.Blob.Size == 0 { + return fmt.Errorf("uploaded blob has size 0") + } + var profileRecord comatproto.RepoGetRecord_Output + err = client.Do(ctx, xrpc.Query, "application/json", "com.atproto.repo.getRecord", map[string]any{ + "repo": repoDID, + "collection": "app.bsky.actor.profile", + "rkey": "self", + }, nil, &profileRecord) + if err != nil { + return fmt.Errorf("failed to get profile record: %w", err) + } + actorProfile, ok := profileRecord.Value.Val.(*bsky.ActorProfile) + if !ok { + return fmt.Errorf("profile record is not a actor profile") + } + actorProfile.Avatar = out.Blob + inp := comatproto.RepoPutRecord_Input{ + Collection: "app.bsky.actor.profile", + Record: &lexutil.LexiconTypeDecoder{Val: actorProfile}, + Rkey: "self", + Repo: repoDID, + SwapRecord: profileRecord.Cid, + } + putOutput := comatproto.RepoPutRecord_Output{} + err = client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", map[string]any{}, inp, &putOutput) + if err != nil { + return fmt.Errorf("failed to put profile record: %w", err) + } + log.Log(ctx, "uploaded redcircle", "ref", out.Blob.Ref.String(), "size", out.Blob.Size) + return nil +} diff --git a/pkg/spxrpc/og.go b/pkg/spxrpc/og.go index 9211d1991..a51b1e8fe 100644 --- a/pkg/spxrpc/og.go +++ b/pkg/spxrpc/og.go @@ -20,8 +20,6 @@ import ( "golang.org/x/image/draw" - "github.com/bluesky-social/indigo/api/bsky" - "github.com/bluesky-social/indigo/xrpc" "github.com/labstack/echo/v4" "github.com/patrickmn/go-cache" "github.com/tdewolff/canvas" @@ -90,8 +88,6 @@ const ( descriptionTruncate = 117 ) -var ErrUserNotFound = errors.New("user not found") - // blendWithBackground creates a pseudo-transparent color by blending the given color with the background // alpha should be between 0.0 (fully background) and 1.0 (fully foreground color) func blendWithBackground(fg color.RGBA, bg color.RGBA, alpha float64) color.RGBA { @@ -235,7 +231,7 @@ func (s *Server) generateOGImage(ctx context.Context, username string) ([]byte, handle = username description = "Live streaming platform for creators and their communities." - profileData, err := s.fetchUserProfile(ctx, username) + profileData, err := s.ATSync.FetchUserProfile(ctx, username) if err != nil { return nil, fmt.Errorf("failed to fetch profile, because %w", err) } else if profileData != nil { @@ -513,35 +509,3 @@ func getAtkinsonBold() ([]byte, error) { return data, nil } - -func (s *Server) fetchUserProfile(ctx context.Context, username string) (*bsky.ActorDefs_ProfileViewDetailed, error) { - // Use ATSync to resolve username to DID, then fetch full profile from Bluesky - var actor string - - // First try to resolve via internal DB - repo, err := s.ATSync.Model.GetRepoByHandleOrDID(username) - if err != nil { - return nil, fmt.Errorf("%w: %w", ErrUserNotFound, err) - } else if repo != nil { - // Use the DID as it's the most reliable identifier - actor = repo.DID - } else { - return nil, fmt.Errorf("no repo found for username: %s (%w)", username, ErrUserNotFound) - } - - // Fetch full profile from Bluesky public API - client := &xrpc.Client{ - Host: "https://public.api.bsky.app", - } - - profile, err := bsky.ActorGetProfile(ctx, client, actor) - if err != nil { - return nil, fmt.Errorf("failed to fetch profile from Bluesky for '%s': %w", actor, err) - } - - if profile == nil { - return nil, fmt.Errorf("received nil profile from Bluesky API for '%s'", actor) - } - - return profile, nil -}