add hotness queries and periodic materializer (#25)

also reformatted some sql

.env.example

@@ -1,6 +1,7 @@
 ENV=dev
 BFF_INGESTER_ENABLED=1
 BFF_API_ENABLED=1
+BFF_HOTNESS_MATERIALIZER_ENABLED=1
 
 # Your handle on bsky.app and an app password generated
 # at: https://bsky.app/settings/app-passwords

cmd/bffsrv/main.go

@@ -12,6 +12,7 @@ import (
 	"github.com/strideynet/bsky-furry-feed/api"
 	"github.com/strideynet/bsky-furry-feed/bluesky"
 	"github.com/strideynet/bsky-furry-feed/feed"
+	"github.com/strideynet/bsky-furry-feed/hotness"
 	"github.com/strideynet/bsky-furry-feed/ingester"
 	"github.com/strideynet/bsky-furry-feed/store"
 	"go.opentelemetry.io/contrib/detectors/gcp"
@@ -113,6 +114,7 @@ func runE(log *zap.Logger) error {
 
 	ingesterEnabled := os.Getenv("BFF_INGESTER_ENABLED") == "1"
 	apiEnabled := os.Getenv("BFF_API_ENABLED") == "1"
+	hotnessMaterializerEnabled := os.Getenv("BFF_HOTNESS_MATERIALIZER_ENABLED") == "1"
 
 	log.Info("starting", zap.String("mode", string(mode)))
 
@@ -227,6 +229,17 @@ func runE(log *zap.Logger) error {
 		})
 	}
 
+	if hotnessMaterializerEnabled {
+		hm := hotness.NewMaterializer(log.Named("hotness"), pgxStore, hotness.Opts{
+			MaterializationPeriod: 30 * time.Second,
+			RetentionPeriod:       1 * time.Hour,
+			LookbackPeriod:        24 * time.Hour,
+		})
+		eg.Go(func() error {
+			return hm.Run(ctx)
+		})
+	}
+
 	// Setup private diagnostics/metrics server
 	debugSrv := debugServer()
 	eg.Go(func() error {

hotness/materializer.go

@@ -0,0 +1,67 @@
+package hotness
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/strideynet/bsky-furry-feed/store"
+	"go.uber.org/zap"
+)
+
+type Materializer struct {
+	log   *zap.Logger
+	store *store.PGXStore
+	opts  Opts
+}
+
+type Opts struct {
+	MaterializationPeriod time.Duration
+	RetentionPeriod       time.Duration
+	LookbackPeriod        time.Duration
+}
+
+func NewMaterializer(
+	log *zap.Logger, store *store.PGXStore, opts Opts,
+) *Materializer {
+	return &Materializer{
+		log:   log,
+		store: store,
+		opts:  opts,
+	}
+}
+
+func (m *Materializer) materialize(ctx context.Context) error {
+	return m.store.MaterializeClassicPostHotness(ctx, m.opts.LookbackPeriod)
+}
+
+func (m *Materializer) cleanup(ctx context.Context) error {
+	return m.store.DeleteOldPostHotness(ctx, m.opts.RetentionPeriod)
+}
+
+func (m *Materializer) step(ctx context.Context) error {
+	// NOTE: materalize and cleanup don't run a transaction together (they don't need to, since it's okay if we keep old materialized results around for too long).
+	// However, we should do the cleanup _after_ the materialization, in case the materialization fails and we converge on purging the entire table.
+	if err := m.materialize(ctx); err != nil {
+		return fmt.Errorf("materialize: %w", err)
+	}
+	if err := m.cleanup(ctx); err != nil {
+		return fmt.Errorf("cleanup: %w", err)
+	}
+	return nil
+}
+
+func (m *Materializer) Run(ctx context.Context) error {
+	t := time.NewTicker(m.opts.MaterializationPeriod)
+	for {
+		select {
+		case <-t.C:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+
+		if err := m.step(ctx); err != nil {
+			m.log.Error("step", zap.Error(err))
+		}
+	}
+}

sqlc.yaml

@@ -7,6 +7,7 @@ overrides:
        actor_did: ActorDID
        subject_uri: SubjectURI
        is_nsfw: IsNSFW
+       after_uri: AfterURI
      overrides:
        - column: candidate_posts.raw
          go_type:

store/migrations/000017_create_post_hotness.down.sql

@@ -0,0 +1,2 @@
+DROP SEQUENCE post_hotness_generation_seq;
+DROP TABLE post_hotness;