diff options
| -rw-r--r-- | env.example | 5 | ||||
| -rw-r--r-- | fetchers/cache.go | 39 | ||||
| -rw-r--r-- | fetchers/jetstream.go | 6 | ||||
| -rw-r--r-- | main.go | 28 |
4 files changed, 51 insertions, 27 deletions
diff --git a/env.example b/env.example index 2990223..5dcc174 100644 --- a/env.example +++ b/env.example @@ -11,8 +11,9 @@ SLINGSHOT_URL=https://slingshot.microcosm.blue VALKEY_ADDR=valkey:6379 VALKEY_PASSWORD= -# cache TTL in minutes -TTL_MIN=4320 +# cache TTL in minutes: posts default 3 days, profiles default 7 days +POST_TTL_MIN=4320 +PROFILE_TTL_MIN=10080 # porxie blob proxy PORXIE_ADDR=porxie:6314 diff --git a/fetchers/cache.go b/fetchers/cache.go index 4196792..3802823 100644 --- a/fetchers/cache.go +++ b/fetchers/cache.go @@ -12,12 +12,13 @@ import ( ) type Cache struct { - client *redis.Client - ctx context.Context - expiryTime time.Duration + client *redis.Client + ctx context.Context + postExpiry time.Duration + profileExpiry time.Duration } -func NewCache(addr, password string, expiryMinutes int) (*Cache, error) { +func NewCache(addr, password string, postExpiryMinutes, profileExpiryMinutes int) (*Cache, error) { client := redis.NewClient(&redis.Options{ Addr: addr, Password: password, @@ -30,9 +31,10 @@ func NewCache(addr, password string, expiryMinutes int) (*Cache, error) { } return &Cache{ - client: client, - ctx: ctx, - expiryTime: time.Duration(expiryMinutes) * time.Minute, + client: client, + ctx: ctx, + postExpiry: time.Duration(postExpiryMinutes) * time.Minute, + profileExpiry: time.Duration(profileExpiryMinutes) * time.Minute, }, nil } @@ -48,6 +50,7 @@ func (c *Cache) Get(uri string) (*notes.Note, bool) { return nil, false } + c.client.Expire(c.ctx, key, c.postExpiry) return ¬e, true } @@ -58,7 +61,7 @@ func (c *Cache) Set(uri string, note *notes.Note) { return } - c.client.Set(c.ctx, key, data, c.expiryTime) + c.client.Set(c.ctx, key, data, c.postExpiry) } func (c *Cache) Delete(uri string) { @@ -71,7 +74,17 @@ func (c *Cache) DeleteAuthor(did string) { c.client.Del(c.ctx, key) } +// GetAuthor returns a cached author and refreshes its TTL. func (c *Cache) GetAuthor(did string) (*notes.Author, bool) { + return c.getAuthor(did, true) +} + +// readAuthor is the same as GetAuthor but doesn't refresh the TTL. +func (c *Cache) readAuthor(did string) (*notes.Author, bool) { + return c.getAuthor(did, false) +} + +func (c *Cache) getAuthor(did string, refresh bool) (*notes.Author, bool) { key := fmt.Sprintf("profile:%s", did) data, err := c.client.Get(c.ctx, key).Bytes() if err != nil { @@ -83,6 +96,10 @@ func (c *Cache) GetAuthor(did string) (*notes.Author, bool) { return nil, false } + if refresh { + c.client.Expire(c.ctx, key, c.profileExpiry) + } + return &author, true } @@ -93,7 +110,7 @@ func (c *Cache) SetAuthor(did string, author *notes.Author) { return } - c.client.Set(c.ctx, key, data, c.expiryTime) + c.client.Set(c.ctx, key, data, c.profileExpiry) } func (c *Cache) GetCursor(did string) (string, bool) { @@ -120,7 +137,8 @@ func (c *Cache) GetPostsByDID(did string) ([]notes.Note, error) { iter := c.client.Scan(c.ctx, 0, pattern, 0).Iterator() for iter.Next(c.ctx) { - data, err := c.client.Get(c.ctx, iter.Val()).Bytes() + key := iter.Val() + data, err := c.client.Get(c.ctx, key).Bytes() if err != nil { continue } @@ -128,6 +146,7 @@ func (c *Cache) GetPostsByDID(did string) ([]notes.Note, error) { if err := json.Unmarshal(data, ¬e); err != nil { continue } + c.client.Expire(c.ctx, key, c.postExpiry) posts = append(posts, note) } diff --git a/fetchers/jetstream.go b/fetchers/jetstream.go index 78a2550..866533a 100644 --- a/fetchers/jetstream.go +++ b/fetchers/jetstream.go @@ -103,7 +103,7 @@ func handleJetstreamPost(conf *Config, did, uri string, commit *jetstreamCommit) } // only track posts from authors we already care about - author, found := conf.Cache.GetAuthor(did) + author, found := conf.Cache.readAuthor(did) if !found { return } @@ -118,13 +118,13 @@ func handleJetstreamPost(conf *Config, did, uri string, commit *jetstreamCommit) func handleJetstreamProfile(conf *Config, did string, commit *jetstreamCommit) { if commit.Operation == "delete" { - if _, found := conf.Cache.GetAuthor(did); found { + if _, found := conf.Cache.readAuthor(did); found { conf.Cache.DeleteAuthor(did) } return } - existing, found := conf.Cache.GetAuthor(did) + existing, found := conf.Cache.readAuthor(did) if !found { return } @@ -63,21 +63,12 @@ func init() { logger.Debug("Logging configured", "log level", logLevel) slingshotURL := os.Getenv("SLINGSHOT_URL") - ttlMin := os.Getenv("TTL_MIN") + postExpiryMinutes := parseTTLMinutes("POST_TTL_MIN", 4320) + profileExpiryMinutes := parseTTLMinutes("PROFILE_TTL_MIN", 10080) - var expiryMinutes int - if ttlMin != "" { - var err error - expiryMinutes, err = strconv.Atoi(ttlMin) - if err != nil { - panic(err) - } - } else { - expiryMinutes = 15 - } valkeyAddr := os.Getenv("VALKEY_ADDR") valkeyPassword := os.Getenv("VALKEY_PASSWORD") - postCache, err := fetchers.NewCache(valkeyAddr, valkeyPassword, expiryMinutes) + postCache, err := fetchers.NewCache(valkeyAddr, valkeyPassword, postExpiryMinutes, profileExpiryMinutes) if err != nil { logger.Error("failed to connect to valkey", "error", err) panic(err) @@ -103,6 +94,19 @@ func init() { } } +func parseTTLMinutes(envVar string, defaultMinutes int) int { + raw := os.Getenv(envVar) + if raw == "" { + return defaultMinutes + } + + minutes, err := strconv.Atoi(raw) + if err != nil { + panic(err) + } + return minutes +} + func loginHandler(w http.ResponseWriter, r *http.Request) { handle := r.URL.Query().Get("handle") if handle == "" { |