From e24630ef68026bdef23d4571ec36f1d658d71a40 Mon Sep 17 00:00:00 2001 From: Will Andrews Date: Mon, 18 Nov 2024 21:19:34 +0000 Subject: [PATCH] more refactoring --- consumer.go | 35 ++++++++++++++++---------- feed.go => feedgenerator.go | 15 ++++++++---- main.go | 49 +++++++++++++++++++------------------ server.go | 2 +- store/feed.go | 28 ++++++++++----------- 5 files changed, 72 insertions(+), 57 deletions(-) rename feed.go => feedgenerator.go (77%) diff --git a/consumer.go b/consumer.go index 271602a..25aeb31 100644 --- a/consumer.go +++ b/consumer.go @@ -14,14 +14,22 @@ import ( "github.com/bluesky-social/jetstream/pkg/client/schedulers/sequential" "github.com/bluesky-social/jetstream/pkg/models" "github.com/bugsnag/bugsnag-go/v2" - "github.com/willdot/bskyfeedgen/store" ) +type ConsumerStore interface { + GetSubscriptionsForPost(postURI string) ([]string, error) + AddSubscriptionForPost(subscribedPostURI, userDid, subscriptionPostRkey string) error + GetSubscribedPostURI(userDID, subscriptionPostRkey string) (string, error) + DeleteSubscriptionForUser(userDID, postURI string) error + DeleteFeedPostsForSubscribedPostURIandUserDID(subscribedPostURI, userDID string) error +} + type consumer struct { - cfg *client.ClientConfig + cfg *client.ClientConfig + store ConsumerStore } -func NewConsumer(jsAddr string) *consumer { +func NewConsumer(jsAddr string, store ConsumerStore) *consumer { cfg := client.DefaultClientConfig() if jsAddr != "" { cfg.WebsocketURL = jsAddr @@ -31,22 +39,21 @@ func NewConsumer(jsAddr string) *consumer { } cfg.WantedDids = []string{} return &consumer{ - cfg: cfg, + cfg: cfg, + store: store, } } func (con *consumer) Consume(ctx context.Context, feedGen *FeedGenerator, logger *slog.Logger) error { h := &handler{ - seenSeqs: make(map[int64]struct{}), feedGenerator: feedGen, - store: *feedGen.store, + store: con.store, } scheduler := sequential.NewScheduler("jetstream_localdev", logger, h.HandleEvent) defer scheduler.Shutdown() - // TODO: logger - c, err := client.NewClient(con.cfg, slog.Default(), scheduler) + c, err := client.NewClient(con.cfg, logger, scheduler) if err != nil { return fmt.Errorf("failed to create client: %w", err) } @@ -62,10 +69,8 @@ func (con *consumer) Consume(ctx context.Context, feedGen *FeedGenerator, logger } type handler struct { - seenSeqs map[int64]struct{} - highwater int64 feedGenerator *FeedGenerator - store store.Store + store ConsumerStore } func (h *handler) HandleEvent(ctx context.Context, event *models.Event) error { @@ -102,7 +107,11 @@ func (h *handler) handleCreateEvent(_ context.Context, event *models.Event) erro subscribedPostURI := post.Reply.Parent.Uri // look for posts that are "subscribe" so that we can add the post URI to a list of posts we want to find replies for - if strings.Contains(post.Text, "/subscribe") && event.Did == "did:plc:dadhhalkfcq3gucaq25hjqon" { + if strings.Contains(post.Text, "/subscribe") { + // For now just look for me + if event.Did != "did:plc:dadhhalkfcq3gucaq25hjqon" { + return nil + } slog.Info("a post that's subscribing to another post. Adding to posts to look for", "subscribed post URI", subscribedPostURI) return h.addDidToSubscribedPost(subscribedPostURI, event.Did, event.Commit.RKey) } @@ -138,7 +147,7 @@ func (h *handler) handleDeleteEvent(_ context.Context, event *models.Event) erro // delete from feeds for the subscribedPostURI and the users DID first. This is so that if this fails, it can be tried again and the // subscription will be still there - err = h.store.DeleteFeedItemsForSubscribedPostURIandUserDID(subscribedPostURI, event.Did) + err = h.store.DeleteFeedPostsForSubscribedPostURIandUserDID(subscribedPostURI, event.Did) if err != nil { slog.Error("delete feed items for subscribedPostURI and user", "error", err, "subscribedPostURI", subscribedPostURI, "user DID", event.Did) return fmt.Errorf("delete feed items for subscribedPostURI and user: %w", err) diff --git a/feed.go b/feedgenerator.go similarity index 77% rename from feed.go rename to feedgenerator.go index 257aa83..e19f95e 100644 --- a/feed.go +++ b/feedgenerator.go @@ -9,11 +9,16 @@ import ( "github.com/willdot/bskyfeedgen/store" ) +type feedStore interface { + AddFeedPost(feedItem store.FeedPost) error + GetUsersFeed(usersDID string) ([]store.FeedPost, error) +} + type FeedGenerator struct { - store *store.Store + store feedStore } -func NewFeedGenerator(store *store.Store) *FeedGenerator { +func NewFeedGenerator(store feedStore) *FeedGenerator { return &FeedGenerator{ store: store, } @@ -24,7 +29,7 @@ func (f *FeedGenerator) GetFeed(ctx context.Context, userDID, feed, cursor strin Feed: make([]FeedItem, 0, 0), } - usersFeed, err := f.store.GetUsersFeedItems(userDID) + usersFeed, err := f.store.GetUsersFeed(userDID) if err != nil { return resp, fmt.Errorf("get users feed items from DB: %w", err) } @@ -44,12 +49,12 @@ func (f *FeedGenerator) GetFeed(ctx context.Context, userDID, feed, cursor strin func (f *FeedGenerator) AddToFeedPosts(usersDids []string, subscribedPostURI, replyPostURI string) { for _, did := range usersDids { - feedItem := store.FeedItem{ + feedItem := store.FeedPost{ ReplyURI: replyPostURI, UserDID: did, SubscribedPostURI: subscribedPostURI, } - err := f.store.AddFeedItem(feedItem) + err := f.store.AddFeedPost(feedItem) if err != nil { slog.Error("add users feed item", "error", err, "did", did, "reply post URI", replyPostURI) bugsnag.Notify(err) diff --git a/main.go b/main.go index c429177..a8357c3 100644 --- a/main.go +++ b/main.go @@ -3,7 +3,6 @@ package main import ( "context" "errors" - "fmt" "log/slog" "os" "os/signal" @@ -25,10 +24,26 @@ func main() { signals := make(chan os.Signal, 1) signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + enableJS := os.Getenv("ENABLE_JETSTREAM") bugsnagAPIKey := os.Getenv("BUGSNAG_API_KEY") + + feedDidBase := os.Getenv("FEED_DID_BASE") + if feedDidBase == "" { + slog.Error("FEED_DID_BASE not set") + os.Exit(1) + } + feedHost := os.Getenv("FEED_HOST_NAME") + if feedHost == "" { + slog.Error("FEED_HOST_NAME not set") + os.Exit(1) + } + dbMountPath := os.Getenv("RAILWAY_VOLUME_MOUNT_PATH") + if dbMountPath == "" { + slog.Error("RAILWAY_VOLUME_MOUNT_PATH env not set") + os.Exit(1) + } + if bugsnagAPIKey != "" { bugsnag.Configure(bugsnag.Configuration{ APIKey: bugsnagAPIKey, @@ -36,16 +51,11 @@ func main() { // The import paths for the Go packages containing your source files ProjectPackages: []string{"main", "github.com/willdot/bskyfeedgen"}, // more configuration options + AutoCaptureSessions: false, }) } - dbMountPath := os.Getenv("RAILWAY_VOLUME_MOUNT_PATH") - if dbMountPath == "" { - bugsnag.Notify(fmt.Errorf("RAILWAY_VOLUME_MOUNT_PATH env not set")) - return - } dbFilename := path.Join(dbMountPath, "database.db") - store, err := store.New(dbFilename) if err != nil { slog.Error("create new store", "error", err) @@ -56,21 +66,12 @@ func main() { feeder := NewFeedGenerator(store) - feedDidBase := os.Getenv("FEED_DID_BASE") - if feedDidBase == "" { - slog.Error("FEED_DID_BASE not set") - os.Exit(1) - } - feedHost := os.Getenv("FEED_HOST_NAME") - if feedHost == "" { - slog.Error("FEED_HOST_NAME not set") - os.Exit(1) - } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() - enableJS := os.Getenv("ENABLE_JETSTREAM") if enableJS == "true" { slog.Info("enabling jetstream consume") - go consumeLoop(ctx, jsServerAddr, feeder) + go consumeLoop(ctx, jsServerAddr, feeder, store) } server := NewServer(443, feeder, feedHost, feedDidBase) @@ -86,8 +87,8 @@ func main() { time.Sleep(time.Second) } -func consumeLoop(ctx context.Context, jsServerAddr string, feeder *FeedGenerator) { - consumer := NewConsumer(jsServerAddr) +func consumeLoop(ctx context.Context, jsServerAddr string, feeder *FeedGenerator, store *store.Store) { + consumer := NewConsumer(jsServerAddr, store) retry.Do(func() error { err := consumer.Consume(ctx, feeder, slog.Default()) @@ -99,5 +100,5 @@ func consumeLoop(ctx context.Context, jsServerAddr string, feeder *FeedGenerator return err } return nil - }, retry.Attempts(0)) + }, retry.Attempts(0)) // retry indefinitly until context canceled } diff --git a/server.go b/server.go index 44623ed..05292e6 100644 --- a/server.go +++ b/server.go @@ -70,7 +70,7 @@ func (s *Server) HandleGetFeedSkeleton(w http.ResponseWriter, r *http.Request) { feed := params.Get("feed") if feed == "" { - slog.Error("missing query param", "host", r.RemoteAddr) + slog.Error("missing feed query param", "host", r.RemoteAddr) http.Error(w, "missing feed query param", http.StatusBadRequest) return } diff --git a/store/feed.go b/store/feed.go index 81deb50..d056d5a 100644 --- a/store/feed.go +++ b/store/feed.go @@ -29,55 +29,55 @@ func createFeedTable(db *sql.DB) error { return nil } -type FeedItem struct { +type FeedPost struct { ID int ReplyURI string UserDID string SubscribedPostURI string } -func (s *Store) AddFeedItem(feedItem FeedItem) error { +func (s *Store) AddFeedPost(feedPost FeedPost) error { sql := `INSERT INTO feed (replyURI, userDID, subscribedPostURI) VALUES (?, ?, ?) ON CONFLICT(replyURI, userDID) DO NOTHING;` - _, err := s.db.Exec(sql, feedItem.ReplyURI, feedItem.UserDID, feedItem.SubscribedPostURI) + _, err := s.db.Exec(sql, feedPost.ReplyURI, feedPost.UserDID, feedPost.SubscribedPostURI) if err != nil { return fmt.Errorf("exec insert feed item: %w", err) } return nil } -func (s *Store) GetUsersFeedItems(usersDID string) ([]FeedItem, error) { +func (s *Store) GetUsersFeed(usersDID string) ([]FeedPost, error) { sql := "SELECT id, replyURI, userDID FROM feed WHERE userDID = ?;" rows, err := s.db.Query(sql, usersDID) if err != nil { - return nil, fmt.Errorf("run query to get users feed item: %w", err) + return nil, fmt.Errorf("run query to get users feed posts: %w", err) } defer rows.Close() - feedItems := make([]FeedItem, 0) + feedPosts := make([]FeedPost, 0) for rows.Next() { - var feedItem FeedItem - if err := rows.Scan(&feedItem.ID, &feedItem.ReplyURI, &feedItem.UserDID); err != nil { + var feedPost FeedPost + if err := rows.Scan(&feedPost.ID, &feedPost.ReplyURI, &feedPost.UserDID); err != nil { return nil, fmt.Errorf("scan row: %w", err) } - feedItems = append(feedItems, feedItem) + feedPosts = append(feedPosts, feedPost) } - return feedItems, nil + return feedPosts, nil } -func (s *Store) DeleteFeedItemsForSubscribedPostURIandUserDID(subscribedPostURI, userDID string) error { +func (s *Store) DeleteFeedPostsForSubscribedPostURIandUserDID(subscribedPostURI, userDID string) error { sql := "DELETE FROM feed WHERE subscribedPostURI = ? AND userDID = ?;" statement, err := s.db.Prepare(sql) if err != nil { - return fmt.Errorf("prepare delete feed items: %w", err) + return fmt.Errorf("prepare delete feed posts: %w", err) } res, err := statement.Exec(subscribedPostURI, userDID) if err != nil { - return fmt.Errorf("exec delete feed items: %w", err) + return fmt.Errorf("exec delete feed posts: %w", err) } n, _ := res.RowsAffected() - slog.Info("delete feed res", "affected rows", n) + slog.Info("delete feed posts result", "affected rows", n) return nil } -- 2.51.2