Added jetstream monitoring

7 files changed, 164 insertions, 1 deletions

diff --git a/env.example b/env.example
index 915e70f..2990223 100644
--- a/env.example
+++ b/env.example
@@ -12,7 +12,10 @@ VALKEY_ADDR=valkey:6379
 VALKEY_PASSWORD=
 
 # cache TTL in minutes
-TTL_MIN=15
+TTL_MIN=4320
 
 # porxie blob proxy
 PORXIE_ADDR=porxie:6314
+
+# jetstream instance for live cache invalidation
+JETSTREAM_URL=wss://jetstream2.fr.hose.cam/subscribe
diff --git a/fetchers/cache.go b/fetchers/cache.go
index 11a43f3..4196792 100644
--- a/fetchers/cache.go
+++ b/fetchers/cache.go
@@ -61,6 +61,16 @@ func (c *Cache) Set(uri string, note *notes.Note) {
 	c.client.Set(c.ctx, key, data, c.expiryTime)
 }
 
+func (c *Cache) Delete(uri string) {
+	key := fmt.Sprintf("post:%s", uri)
+	c.client.Del(c.ctx, key)
+}
+
+func (c *Cache) DeleteAuthor(did string) {
+	key := fmt.Sprintf("profile:%s", did)
+	c.client.Del(c.ctx, key)
+}
+
 func (c *Cache) GetAuthor(did string) (*notes.Author, bool) {
 	key := fmt.Sprintf("profile:%s", did)
 	data, err := c.client.Get(c.ctx, key).Bytes()
diff --git a/fetchers/config.go b/fetchers/config.go
index a8aa711..a3a1ccf 100644
--- a/fetchers/config.go
+++ b/fetchers/config.go
@@ -5,6 +5,7 @@ import "log/slog"
 type Config struct {
 	SlingshotURL string
 	PorxieURL    string
+	JetstreamURL string
 	PageSize     int
 	Logger       *slog.Logger
 	Cache        *Cache
diff --git a/fetchers/jetstream.go b/fetchers/jetstream.go
new file mode 100644
index 0000000..78a2550
--- /dev/null
+++ b/fetchers/jetstream.go
@@ -0,0 +1,138 @@
+package fetchers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"llostre/general"
+	"net/url"
+	"time"
+
+	"github.com/coder/websocket"
+)
+
+type jetstreamEvent struct {
+	DID    string           `json:"did"`
+	Kind   string           `json:"kind"`
+	Commit *jetstreamCommit `json:"commit,omitempty"`
+}
+
+type jetstreamCommit struct {
+	Operation  string          `json:"operation"`
+	Collection string          `json:"collection"`
+	RKey       string          `json:"rkey"`
+	Record     json.RawMessage `json:"record,omitempty"`
+}
+
+var jetstreamCollections = []string{"app.bsky.feed.post", "app.bsky.actor.profile"}
+
+// Meant to be run in its own goroutine; blocks forever, reconnecting on failure.
+// Monitors the jetstream for changes in cached records
+func StartJetstream(conf *Config) {
+	if conf.JetstreamURL == "" {
+		return
+	}
+
+	subscribeURL := buildJetstreamURL(conf.JetstreamURL)
+
+	for {
+		if err := consumeJetstream(conf, subscribeURL); err != nil {
+			conf.Logger.Error("jetstream connection lost, reconnecting", "error", err)
+		}
+		time.Sleep(5 * time.Second)
+	}
+}
+
+func buildJetstreamURL(base string) string {
+	query := url.Values{}
+	for _, collection := range jetstreamCollections {
+		query.Add("wantedCollections", collection)
+	}
+	return fmt.Sprintf("%s?%s", base, query.Encode())
+}
+
+func consumeJetstream(conf *Config, subscribeURL string) error {
+	ctx := context.Background()
+
+	conn, _, err := websocket.Dial(ctx, subscribeURL, nil)
+	if err != nil {
+		return err
+	}
+	defer conn.CloseNow()
+	conn.SetReadLimit(1 << 20) // 1MiB; default 32KiB is too tight for larger posts/profiles
+
+	conf.Logger.Debug("jetstream connected", "url", subscribeURL)
+
+	for {
+		_, data, err := conn.Read(ctx)
+		if err != nil {
+			return err
+		}
+
+		var event jetstreamEvent
+		if err := json.Unmarshal(data, &event); err != nil {
+			conf.Logger.Warn("failed to parse jetstream event", "error", err)
+			continue
+		}
+
+		if event.Kind != "commit" || event.Commit == nil {
+			continue
+		}
+
+		handleJetstreamCommit(conf, event.DID, event.Commit)
+	}
+}
+
+func handleJetstreamCommit(conf *Config, did string, commit *jetstreamCommit) {
+	uri := fmt.Sprintf("at://%s/%s/%s", did, commit.Collection, commit.RKey)
+
+	switch commit.Collection {
+	case "app.bsky.feed.post":
+		handleJetstreamPost(conf, did, uri, commit)
+	case "app.bsky.actor.profile":
+		handleJetstreamProfile(conf, did, commit)
+	}
+}
+
+func handleJetstreamPost(conf *Config, did, uri string, commit *jetstreamCommit) {
+	if commit.Operation == "delete" {
+		if _, found := conf.Cache.Get(uri); found {
+			conf.Cache.Delete(uri)
+		}
+		return
+	}
+
+	// only track posts from authors we already care about
+	author, found := conf.Cache.GetAuthor(did)
+	if !found {
+		return
+	}
+
+	rawRecord := &general.RawRecord{URI: uri, Value: commit.Record}
+
+	// cache the result
+	if _, err := parseAppBskyFeedPostRecord(rawRecord, author, conf); err != nil {
+		conf.Logger.Warn("jetstream: failed to parse post", "uri", uri, "error", err)
+	}
+}
+
+func handleJetstreamProfile(conf *Config, did string, commit *jetstreamCommit) {
+	if commit.Operation == "delete" {
+		if _, found := conf.Cache.GetAuthor(did); found {
+			conf.Cache.DeleteAuthor(did)
+		}
+		return
+	}
+
+	existing, found := conf.Cache.GetAuthor(did)
+	if !found {
+		return
+	}
+
+	rawRecord := &general.RawRecord{URI: fmt.Sprintf("at://%s/app.bsky.actor.profile/self", did), Value: commit.Record}
+
+	// cache the result
+	if _, err := parseAppBskyActorProfile(did, existing.PDS, rawRecord, conf.Cache); err != nil {
+		conf.Logger.Warn("jetstream: failed to parse updated profile", "did", did, "error", err)
+	}
+}
diff --git a/go.mod b/go.mod
index fc20cdc..86f5e03 100644
--- a/go.mod
+++ b/go.mod
@@ -3,6 +3,7 @@ module llostre
 go 1.26.5
 
 require (
+	github.com/coder/websocket v1.8.15
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/joho/godotenv v1.5.1
 	github.com/redis/go-redis/v9 v9.18.0
diff --git a/go.sum b/go.sum
index f29f86b..b0e57b3 100644
--- a/go.sum
+++ b/go.sum
@@ -4,6 +4,8 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
+github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
diff --git a/main.go b/main.go
index 17eee09..adb0f91 100644
--- a/main.go
+++ b/main.go
@@ -88,9 +88,15 @@ func init() {
 		porxieAddr = "porxie:6314"
 	}
 
+	jetstreamURL := os.Getenv("JETSTREAM_URL")
+	if jetstreamURL == "" {
+		jetstreamURL = "wss://jetstream2.fr.hose.cam/subscribe"
+	}
+
 	config = &fetchers.Config{
 		SlingshotURL: slingshotURL,
 		PorxieURL:    fmt.Sprintf("http://%s", porxieAddr),
+		JetstreamURL: jetstreamURL,
 		PageSize:     50,
 		Logger:       logger,
 		Cache:        postCache,
@@ -449,6 +455,8 @@ func blobHandler(w http.ResponseWriter, r *http.Request) {
 func main() {
 	defer config.Cache.Close()
 
+	go fetchers.StartJetstream(config)
+
 	// serve static files
 	distFS, err := fs.Sub(embeddedFrontend, "frontend/dist")
 	if err != nil {