Merge pull request #118 from strideynet/self-labelling-support

Self labelling support for posts and profiles.

docs/operations.md

@@ -26,3 +26,14 @@ care of the "password" element (short lived tokens are injected).
 
 Permissions may be a bit screwy at the moment. You'll need to manually grant
 access to tables to cloudsqliamserviceaccount and cloudsqliamuser.
+
+### Runbook
+
+1. Cut release on GitHub with tag
+2. Wait for docker image build to succeed
+3. If migration is necessary. If so, run it
+4. If migration created tables, assign permissions to the correct groups
+5. Update k8s manifests
+6. Apply manifests
+7. Monitor deployment
+8. Celebrate

feed/feed_test.go

@@ -32,6 +32,7 @@ func TestChronologicalGenerator(t *testing.T) {
 	artPost := indigoTest.RandFakeAtUri("app.bsky.feed.post", "art")
 	nsfwArtPost := indigoTest.RandFakeAtUri("app.bsky.feed.post", "nsfwArt")
 	poastPost := indigoTest.RandFakeAtUri("app.bsky.feed.post", "poast")
+	nsfwLabelledPost := indigoTest.RandFakeAtUri("app.bsky.feed.post", "nsfw-labelled")
 
 	for _, opts := range []store.CreatePostOpts{
 		{
@@ -79,6 +80,16 @@ func TestChronologicalGenerator(t *testing.T) {
 			HasMedia:  true,
 			Raw:       &bsky.FeedPost{},
 		},
+		{
+			URI:        nsfwLabelledPost,
+			ActorDID:   furry.DID(),
+			CreatedAt:  time.Time{},
+			IndexedAt:  time.Time{},
+			Hashtags:   []string{"art"},
+			HasMedia:   true,
+			Raw:        &bsky.FeedPost{},
+			SelfLabels: []string{"sexual"},
+		},
 	} {
 		require.NoError(t, harness.Store.CreatePost(ctx, opts))
 	}
@@ -95,7 +106,14 @@ func TestChronologicalGenerator(t *testing.T) {
 				IsNSFW:   tristate.Maybe,
 				HasMedia: tristate.Maybe,
 			},
-			expectedPosts: []string{fursuitPost, murrsuitPost, artPost, nsfwArtPost, poastPost},
+			expectedPosts: []string{
+				fursuitPost,
+				murrsuitPost,
+				artPost,
+				nsfwArtPost,
+				poastPost,
+				nsfwLabelledPost,
+			},
 		},
 		{
 			name: "all fursuits",
@@ -122,7 +140,7 @@ func TestChronologicalGenerator(t *testing.T) {
 				IsNSFW:   tristate.True,
 				HasMedia: tristate.True,
 			},
-			expectedPosts: []string{nsfwArtPost},
+			expectedPosts: []string{nsfwArtPost, nsfwLabelledPost},
 		},
 	} {
 		test := test

go.mod

@@ -6,7 +6,7 @@ require (
 	cloud.google.com/go/cloudsqlconn v1.2.4
 	connectrpc.com/connect v1.11.0
 	connectrpc.com/otelconnect v0.5.0
-	github.com/bluesky-social/indigo v0.0.0-20230728184334-c2ae70612145
+	github.com/bluesky-social/indigo v0.0.0-20230815172301-d9efabf04191
 	github.com/golang-migrate/migrate/v4 v4.16.2
 	github.com/google/go-cmp v0.5.9
 	github.com/gorilla/websocket v1.5.0

go.sum

@@ -73,6 +73,14 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bluesky-social/indigo v0.0.0-20230728184334-c2ae70612145 h1:La+KOOafHIKEK3WOJRvypdUFU0PVovMdca8wdoXuYQc=
 github.com/bluesky-social/indigo v0.0.0-20230728184334-c2ae70612145/go.mod h1:R4lTJrJ8FkabyT9oqn3nIB+49xqcGPNdlHLq2huREjY=
+github.com/bluesky-social/indigo v0.0.0-20230810233313-ca1f9da51f09 h1:9Jh/ez4Ie3+YCGhI9agCQcB97AIHjwJvXAu/gEzdcP4=
+github.com/bluesky-social/indigo v0.0.0-20230810233313-ca1f9da51f09/go.mod h1:mvd5qjdsWSCYMu8r0Lc+M7FNbBi63mmTMJnPh+F1ghI=
+github.com/bluesky-social/indigo v0.0.0-20230811212902-2bdfb5bc4fdc h1:DTZVw4dC40fPdFtQho+++q6B21X58qMMeJKrq2HLwIg=
+github.com/bluesky-social/indigo v0.0.0-20230811212902-2bdfb5bc4fdc/go.mod h1:mvd5qjdsWSCYMu8r0Lc+M7FNbBi63mmTMJnPh+F1ghI=
+github.com/bluesky-social/indigo v0.0.0-20230815171710-dcd896ab2eb2 h1:gyUHE69cDQc+3DaBfjraoJSH47nIux2z6xt0jRdsSTM=
+github.com/bluesky-social/indigo v0.0.0-20230815171710-dcd896ab2eb2/go.mod h1:mvd5qjdsWSCYMu8r0Lc+M7FNbBi63mmTMJnPh+F1ghI=
+github.com/bluesky-social/indigo v0.0.0-20230815172301-d9efabf04191 h1:1ecZfDDYbqXLHbbCkLYRSZf2c3rpreqyK/bjb5qKlUY=
+github.com/bluesky-social/indigo v0.0.0-20230815172301-d9efabf04191/go.mod h1:mvd5qjdsWSCYMu8r0Lc+M7FNbBi63mmTMJnPh+F1ghI=
 github.com/brianvoe/gofakeit/v6 v6.20.2 h1:FLloufuC7NcbHqDzVQ42CG9AKryS1gAGCRt8nQRsW+Y=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=

ingester/handle_actor_profile.go

@@ -33,6 +33,13 @@ func (fi *FirehoseIngester) handleActorProfileUpdate(
 		description = *data.Description
 	}
 
+	selfLabels := []string{}
+	if data.Labels != nil && data.Labels.LabelDefs_SelfLabels != nil {
+		for _, label := range data.Labels.LabelDefs_SelfLabels.Values {
+			selfLabels = append(selfLabels, label.Val)
+		}
+	}
+
 	err = fi.store.CreateLatestActorProfile(
 		ctx,
 		store.CreateLatestActorProfileOpts{
@@ -42,6 +49,7 @@ func (fi *FirehoseIngester) handleActorProfileUpdate(
 			IndexedAt:   time.Now(),
 			DisplayName: displayName,
 			Description: description,
+			SelfLabels:  selfLabels,
 		},
 	)
 	if err != nil {

ingester/handle_feed_post.go

@@ -66,30 +66,40 @@ func (fi *FirehoseIngester) handleFeedPostCreate(
 		endSpan(span, err)
 	}()
 
-	if data.Reply == nil {
-		createdAt, err := bluesky.ParseTime(data.CreatedAt)
-		if err != nil {
-			return fmt.Errorf("parsing post time: %w", err)
-		}
+	if data.Reply != nil {
+		span.AddEvent("ignoring post as it is a reply")
+		return
+	}
+
+	createdAt, err := bluesky.ParseTime(data.CreatedAt)
+	if err != nil {
+		return fmt.Errorf("parsing post time: %w", err)
+	}
 
-		err = fi.store.CreatePost(
-			ctx,
-			store.CreatePostOpts{
-				URI:       recordUri,
-				ActorDID:  repoDID,
-				CreatedAt: createdAt,
-				IndexedAt: time.Now(),
-				Raw:       data,
-				Hashtags:  extractNormalizedHashtags(data),
-				HasMedia:  hasMedia(data),
-			},
-		)
-		if err != nil {
-			return fmt.Errorf("creating post: %w", err)
+	selfLabels := []string{}
+	if data.Labels != nil && data.Labels.LabelDefs_SelfLabels != nil {
+		for _, label := range data.Labels.LabelDefs_SelfLabels.Values {
+			selfLabels = append(selfLabels, label.Val)
 		}
-	} else {
-		span.AddEvent("ignoring post as it is a reply")
 	}
+
+	err = fi.store.CreatePost(
+		ctx,
+		store.CreatePostOpts{
+			URI:        recordUri,
+			ActorDID:   repoDID,
+			CreatedAt:  createdAt,
+			IndexedAt:  time.Now(),
+			Raw:        data,
+			Hashtags:   extractNormalizedHashtags(data),
+			HasMedia:   hasMedia(data),
+			SelfLabels: selfLabels,
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("creating post: %w", err)
+	}
+
 	return nil
 }
 

ingester/ingester.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/bluesky-social/indigo/events/schedulers/sequential"
 	"strconv"
 
 	"github.com/bluesky-social/indigo/util"
@@ -259,9 +260,8 @@ func (fi *FirehoseIngester) Start(ctx context.Context) error {
 				return nil
 			},
 		}
-		return events.HandleRepoStream(ctx, con, &events.SequentialScheduler{
-			Do: callbacks.EventHandler,
-		})
+		scheduler := sequential.NewScheduler("main", callbacks.EventHandler)
+		return events.HandleRepoStream(ctx, con, scheduler)
 		// TODO: sometimes stream exits of own accord, we should attempt to
 		// reconnect and enter an "error state".
 	})
@@ -282,7 +282,6 @@ func (fi *FirehoseIngester) handleCommit(ctx context.Context, evt *atproto.SyncS
 	if err != nil {
 		return fmt.Errorf("parsing timestamp: %w", err)
 	}
-
 	rr, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.Blocks))
 	if err != nil {
 		return fmt.Errorf("reading repo from car: %w", err)

ingester/ingester_test.go

@@ -115,6 +115,7 @@ func TestFirehoseIngester(t *testing.T) {
 					Bool:  false,
 					Valid: true,
 				},
+				SelfLabels: []string{},
 			},
 		},
 		{
@@ -153,6 +154,7 @@ func TestFirehoseIngester(t *testing.T) {
 					Bool:  true,
 					Valid: true,
 				},
+				SelfLabels: []string{},
 			},
 		},
 		{
@@ -177,6 +179,7 @@ func TestFirehoseIngester(t *testing.T) {
 					Bool:  false,
 					Valid: true,
 				},
+				SelfLabels: []string{},
 			},
 		},
 		{
@@ -215,6 +218,38 @@ func TestFirehoseIngester(t *testing.T) {
 					Bool:  true,
 					Valid: true,
 				},
+				SelfLabels: []string{},
+			},
+		},
+		{
+			name: "self labels",
+			user: approvedFurry,
+			post: &bsky.FeedPost{
+				LexiconTypeID: "app.bsky.feed.post",
+				CreatedAt:     now.Format(time.RFC3339Nano),
+				Text:          "paws paws paws",
+				Labels: &bsky.FeedPost_Labels{
+					LabelDefs_SelfLabels: &atproto.LabelDefs_SelfLabels{
+						Values: []*atproto.LabelDefs_SelfLabel{
+							{
+								Val: "adult",
+							},
+						},
+					},
+				},
+			},
+			wantPost: &gen.CandidatePost{
+				ActorDID: approvedFurry.DID(),
+				CreatedAt: pgtype.Timestamptz{
+					Time:  now,
+					Valid: true,
+				},
+				Hashtags: []string{},
+				HasMedia: pgtype.Bool{
+					Bool:  false,
+					Valid: true,
+				},
+				SelfLabels: []string{},
 			},
 		},
 	}