Chore(cache): refactor/fix s3

- implement expiration (through proper meta data)
- add a goroutine to run in the background

Author: till

cmd/bot/main.go

@@ -124,11 +124,12 @@ func main() {
 				Client: client,
 			}
 
-			cacheClient := &content.CacheClientS3{
-				MC:     mc,
-				Bucket: cacheBucket,
-				CTX:    ctx,
-			}
+			cacheClient := content.NewCacheClientS3(ctx, mc, cacheBucket)
+
+			// Initialize and start the cleanup handler
+			cleanup := content.NewS3Cleanup(ctx, mc, cacheBucket)
+			cleanup.Start()
+			defer cleanup.Stop()
 
 			if err := content.Start(githubToken, cacheClient); err != nil {
 				return fmt.Errorf("failed to start service: %v", err)

internal/content/content.go

@@ -3,13 +3,15 @@ package content
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"strings"
 
 	"github.com/ezeoleaf/larry/cache"
 	"github.com/ezeoleaf/larry/config"
 	"github.com/ezeoleaf/larry/provider/github"
 	"github.com/till/golangoss-bluesky/internal/bluesky"
+	"github.com/till/golangoss-bluesky/internal/utils"
 )
 
 var (
@@ -29,7 +31,7 @@ func Start(token string, cacheClient cache.Client) error {
 func Do(ctx context.Context, c bluesky.Client) error {
 	p, err := provider.GetContentToPublish()
 	if err != nil {
-		slog.Error("error fetching content", slog.Any("err", err))
+		utils.LogError(fmt.Errorf("error fetching content: %w", err))
 		return ErrCouldNotContent
 	}
 

internal/content/s3_cache.go

@@ -13,21 +13,46 @@ import (
 
 // CacheClientS3 is a small cache that is backed by an S3-compatible store
 type CacheClientS3 struct {
-	MC     *minio.Client
-	Bucket string
-	CTX    context.Context
+	mc                *minio.Client
+	bucket            string
+	ctx               context.Context
+	defaultExpiration time.Duration
 }
 
-func (c *CacheClientS3) Set(key string, value interface{}, exp time.Duration) error {
-	data, err := json.Marshal(value)
-	if err != nil {
+// NewCacheClientS3 creates a new S3 cache client with default settings
+func NewCacheClientS3(ctx context.Context, mc *minio.Client, bucket string) *CacheClientS3 {
+	return &CacheClientS3{
+		mc:                mc,
+		bucket:            bucket,
+		ctx:               ctx,
+		defaultExpiration: 24 * time.Hour,
+	}
+}
+
+func (c *CacheClientS3) Set(key string, value any, exp time.Duration) error {
+	var data bytes.Buffer
+	if err := json.NewEncoder(&data).Encode(value); err != nil {
 		return err
 	}
 
-	r := bytes.NewReader(data)
+	r := bytes.NewReader(data.Bytes())
+
+	// Use the provided expiration time or fall back to default
+	expiration := exp
+	if expiration == 0 {
+		expiration = c.defaultExpiration
+	}
+
+	// Calculate the expiration time
+	expiresAt := time.Now().Add(expiration)
 
-	_, err = c.MC.PutObject(c.CTX, c.Bucket, key, r, int64(r.Len()), minio.PutObjectOptions{
-		Expires: time.Now().Add(exp),
+	// Set metadata to track expiration
+	metadata := map[string]string{
+		"expires-at": expiresAt.Format(time.RFC3339),
+	}
+
+	_, err := c.mc.PutObject(c.ctx, c.bucket, key, r, int64(r.Len()), minio.PutObjectOptions{
+		UserMetadata: metadata,
 	})
 	return err
 }
@@ -36,17 +61,27 @@ func (c *CacheClientS3) Set(key string, value interface{}, exp time.Duration) er
 // does not exist, in other case we can use minio.ToErrorResponse(err) to extract more details about the
 // potential S3 related error
 func (c *CacheClientS3) Get(key string) (string, error) {
-	if _, err := c.MC.StatObject(c.CTX, c.Bucket, key, minio.StatObjectOptions{}); err != nil {
+	// First check if object exists and get its metadata
+	objInfo, err := c.mc.StatObject(c.ctx, c.bucket, key, minio.StatObjectOptions{})
+	if err != nil {
 		return "", redis.Nil
 	}
 
-	object, err := c.MC.GetObject(c.CTX, c.Bucket, key, minio.GetObjectOptions{})
+	if expiresAt, ok := objInfo.UserMetadata["expires-at"]; ok {
+		expTime, err := time.Parse(time.RFC3339, expiresAt)
+		if err == nil && time.Now().After(expTime) {
+			// Object has expired, delete it and return not found
+			_ = c.Del(key) // Ignore delete error
+			return "", redis.Nil
+		}
+	}
+
+	object, err := c.mc.GetObject(c.ctx, c.bucket, key, minio.GetObjectOptions{})
 	if err != nil {
 		return "", err
 	}
 
 	var val any
-
 	if err := json.NewDecoder(object).Decode(&val); err != nil {
 		return "", err
 	}
@@ -63,12 +98,11 @@ func (c *CacheClientS3) Get(key string) (string, error) {
 }
 
 func (c *CacheClientS3) Del(key string) error {
-	return c.MC.RemoveObject(c.CTX, c.Bucket, key, minio.RemoveObjectOptions{
+	return c.mc.RemoveObject(c.ctx, c.bucket, key, minio.RemoveObjectOptions{
 		ForceDelete: true,
 	})
 }
 
 func (c *CacheClientS3) Scan(key string, action func(context.Context, string) error) error {
-
 	return nil
 }

internal/content/s3_cleanup.go

@@ -0,0 +1,110 @@
+package content
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/minio/minio-go/v7"
+	"github.com/till/golangoss-bluesky/internal/utils"
+)
+
+// S3Cleanup handles background cleanup of expired objects in S3
+type S3Cleanup struct {
+	mc     *minio.Client
+	bucket string
+	// cleanupInterval is how often to run the cleanup routine
+	cleanupInterval time.Duration
+	// stopCleanup is used to signal the cleanup routine to stop
+	stopCleanup chan struct{}
+}
+
+// NewS3Cleanup creates a new S3 cleanup handler
+func NewS3Cleanup(mc *minio.Client, bucket string) *S3Cleanup {
+	return &S3Cleanup{
+		mc:              mc,
+		bucket:          bucket,
+		cleanupInterval: 24 * time.Hour,
+		stopCleanup:     make(chan struct{}),
+	}
+}
+
+// Start begins the background cleanup routine
+func (c *S3Cleanup) Start(ctx context.Context) {
+	go c.cleanupRoutine(ctx)
+}
+
+// Stop stops the background cleanup routine
+func (c *S3Cleanup) Stop() {
+	close(c.stopCleanup)
+}
+
+// cleanupRoutine periodically checks for and deletes expired objects
+func (c *S3Cleanup) cleanupRoutine(ctx context.Context) {
+	ticker := time.NewTicker(c.cleanupInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			if err := c.cleanupExpired(ctx); err != nil {
+				utils.LogErrorWithContext(ctx, fmt.Errorf("failed to cleanup expired objects: %w", err))
+			}
+		case <-c.stopCleanup:
+			return
+		}
+	}
+}
+
+// cleanupExpired scans the bucket for expired objects and deletes them
+func (c *S3Cleanup) cleanupExpired(ctx context.Context) error {
+	filter := minio.ListObjectsOptions{
+		Recursive:    true,
+		WithMetadata: true,
+	}
+
+	objectsCh := c.mc.ListObjects(ctx, c.bucket, filter)
+	for obj := range objectsCh {
+		if obj.Err != nil {
+			utils.LogErrorWithContext(ctx, fmt.Errorf("failed to list objects: %w", obj.Err))
+			continue
+		}
+
+		// Check if object has expiration metadata
+		expiresAt, ok := obj.UserMetadata["expires-at"]
+		if !ok {
+			utils.LogErrorWithContext(
+				ctx,
+				fmt.Errorf("no expiration metadata found for object: %s", obj.Key),
+			)
+			if err := c.delete(ctx, obj); err != nil {
+				return err
+			}
+			continue
+		}
+
+		expTime, err := time.Parse(time.RFC3339, expiresAt)
+		if err != nil {
+			utils.LogError(fmt.Errorf("failed to parse expiration time (%s): %w", obj.Key, err))
+			continue
+		}
+		if !time.Now().After(expTime) {
+			continue
+		}
+
+		// Object has expired, delete it
+		if err := c.delete(ctx, obj); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (c *S3Cleanup) delete(ctx context.Context, obj minio.ObjectInfo) error {
+	if err := c.mc.RemoveObject(ctx, c.bucket, obj.Key, minio.RemoveObjectOptions{
+		ForceDelete: true,
+	}); err != nil {
+		return fmt.Errorf("failed to delete expired object (%s): %w", obj.Key, err)
+	}
+	return nil
+}

internal/utils/utils.go

@@ -0,0 +1,18 @@
+package utils
+
+import (
+	"context"
+	"log/slog"
+)
+
+func LogError(err error) {
+	slog.Error(err.Error(), LogAttr(err))
+}
+
+func LogErrorWithContext(ctx context.Context, err error) {
+	slog.ErrorContext(ctx, err.Error(), LogAttr(err))
+}
+
+func LogAttr(err error) slog.Attr {
+	return slog.Any("error", err)
+}