feat(distributed-cache): Add distributed cache support

.changeset/flag-definition-cache-provider.md

@@ -0,0 +1,5 @@
+---
+"posthog-go": minor
+---
+
+Add the experimental `FlagDefinitionCacheProvider` interface and the `Config.FlagDefinitionCacheProvider` option, for sharing local evaluation flag definitions across SDK instances. See [Distributed environments](https://posthog.com/docs/feature-flags/local-evaluation/distributed-environments).

api/public-api.txt

@@ -314,6 +314,8 @@ type Config struct {
 
 	NextFeatureFlagsPollingTick func() time.Duration
 
+	FlagDefinitionCacheProvider FlagDefinitionCacheProvider
+
 	HistoricalMigration bool
 
 	Transport http.RoundTripper
@@ -624,6 +626,24 @@ type Filter struct {
 	VariantLookupTable []FlagVariantMeta `json:"-"`
 }
 
+type FlagDefinitionCacheData struct {
+	Flags                   []FeatureFlag            `json:"flags"`
+	GroupTypeMapping        map[string]string        `json:"group_type_mapping"`
+	Cohorts                 map[string]PropertyGroup `json:"cohorts"`
+	MinimalFlagCalledEvents bool                     `json:"minimal_flag_called_events"`
+}
+
+type FlagDefinitionCacheProvider interface {
+	GetFlagDefinitions(ctx context.Context) (*FlagDefinitionCacheData, error)
+
+	ShouldFetchFlagDefinitions(ctx context.Context) (bool, error)
+
+	OnFlagDefinitionsReceived(ctx context.Context, data FlagDefinitionCacheData) error
+
+	Shutdown(ctx context.Context) error
+}
+
+
 type FlagDetail struct {
 	Key string `json:"key"`
 	Enabled bool `json:"enabled"`

config.go

@@ -126,6 +126,14 @@ type Config struct {
 	// polling delay. When set, it overrides DefaultFeatureFlagsPollingInterval.
 	NextFeatureFlagsPollingTick func() time.Duration
 
+	// FlagDefinitionCacheProvider shares local evaluation flag definitions with other
+	// SDK instances through an external cache. Only used when SecretKey is configured.
+	// See https://posthog.com/docs/feature-flags/local-evaluation/distributed-environments
+	// for guidance.
+	//
+	// EXPERIMENTAL: this API may change in a minor version bump.
+	FlagDefinitionCacheProvider FlagDefinitionCacheProvider
+
 	// HistoricalMigration marks captured batches as historical migration traffic.
 	// See https://posthog.com/docs/migrate for migration guidance.
 	HistoricalMigration bool

examples/README.md

@@ -31,6 +31,7 @@ This will run:
 - Feature flags example
 - Capture events example
 - Capture events with feature flag options example
+- Feature flags definition cache example
 
 ### Prerequisites
 

examples/flag_definition_cache.go

@@ -0,0 +1,205 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/posthog/posthog-go"
+)
+
+// This example implements posthog.FlagDefinitionCacheProvider with a shared directory,
+// so it runs without extra dependencies. A real deployment shares definitions across
+// hosts, so use Redis or something like it there.
+
+const (
+	// Must be longer than the polling interval.
+	flagCacheLockTTL      = 30 * time.Second
+	flagCachePollInterval = 3 * time.Second
+	flagCacheInstances    = 2
+)
+
+// FileFlagCache shares flag definitions between processes on one machine through a
+// directory: one file holds the definitions, another is the leader election lock.
+type FileFlagCache struct {
+	dir        string
+	name       string
+	instanceID string
+}
+
+// NewFileFlagCache creates a provider for the given cache directory. Instances that
+// share definitions pass the same directory and name.
+func NewFileFlagCache(dir, name string) (*FileFlagCache, error) {
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return nil, err
+	}
+	return &FileFlagCache{dir: dir, name: name, instanceID: uuid.NewString()}, nil
+}
+
+func (c *FileFlagCache) cachePath() string { return filepath.Join(c.dir, c.name+".json") }
+func (c *FileFlagCache) lockPath() string  { return filepath.Join(c.dir, c.name+".lock") }
+
+// ShouldFetchFlagDefinitions elects a single instance to poll PostHog by holding a lock
+// that expires.
+//
+// Read-then-write is not atomic on a filesystem, so two instances starting at the same
+// moment can both take the lock. Redis with a Lua script does this atomically.
+func (c *FileFlagCache) ShouldFetchFlagDefinitions(_ context.Context) (bool, error) {
+	holder, expiry, err := c.readLock()
+	if err != nil {
+		return false, err
+	}
+
+	heldByAnother := holder != "" && holder != c.instanceID
+	if heldByAnother && time.Now().Before(expiry) {
+		return false, nil
+	}
+
+	return true, c.writeLock()
+}
+
+func (c *FileFlagCache) GetFlagDefinitions(_ context.Context) (*posthog.FlagDefinitionCacheData, error) {
+	contents, err := os.ReadFile(c.cachePath())
+	if errors.Is(err, os.ErrNotExist) {
+		return nil, nil
+	} else if err != nil {
+		return nil, err
+	}
+
+	var data posthog.FlagDefinitionCacheData
+	if err := json.Unmarshal(contents, &data); err != nil {
+		return nil, err
+	}
+	return &data, nil
+}
+
+// OnFlagDefinitionsReceived publishes definitions, writing to a temporary file and
+// renaming it so that a reader never observes a half-written payload.
+func (c *FileFlagCache) OnFlagDefinitionsReceived(_ context.Context, data posthog.FlagDefinitionCacheData) error {
+	encoded, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+
+	tmp, err := os.CreateTemp(c.dir, c.name+".*.tmp")
+	if err != nil {
+		return err
+	}
+	defer os.Remove(tmp.Name())
+
+	if _, err := tmp.Write(encoded); err != nil {
+		tmp.Close()
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+
+	fmt.Printf("   [%s] published %d flag definitions to the shared cache\n", c.shortID(), len(data.Flags))
+	return os.Rename(tmp.Name(), c.cachePath())
+}
+
+func (c *FileFlagCache) Shutdown(_ context.Context) error {
+	holder, _, err := c.readLock()
+	if err != nil || holder != c.instanceID {
+		return err
+	}
+	if err := os.Remove(c.lockPath()); err != nil && !errors.Is(err, os.ErrNotExist) {
+		return err
+	}
+	return nil
+}
+
+func (c *FileFlagCache) readLock() (holder string, expiry time.Time, err error) {
+	contents, err := os.ReadFile(c.lockPath())
+	if errors.Is(err, os.ErrNotExist) {
+		return "", time.Time{}, nil
+	} else if err != nil {
+		return "", time.Time{}, err
+	}
+
+	parts := strings.Fields(string(contents))
+	if len(parts) != 2 {
+		return "", time.Time{}, nil
+	}
+	nanos, err := strconv.ParseInt(parts[1], 10, 64)
+	if err != nil {
+		return "", time.Time{}, nil
+	}
+	return parts[0], time.Unix(0, nanos), nil
+}
+
+func (c *FileFlagCache) writeLock() error {
+	expiry := time.Now().Add(flagCacheLockTTL).UnixNano()
+	return os.WriteFile(c.lockPath(), []byte(fmt.Sprintf("%s %d", c.instanceID, expiry)), 0o644)
+}
+
+func (c *FileFlagCache) shortID() string { return c.instanceID[:8] }
+
+// TestFlagDefinitionCache runs several clients that share one flag definition cache.
+func TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint string) {
+	dir, err := os.MkdirTemp("", "posthog-flag-cache")
+	if err != nil {
+		fmt.Printf("❌ Could not create the cache directory: %v\n", err)
+		return
+	}
+	defer os.RemoveAll(dir)
+
+	fmt.Printf("%d instances sharing the cache in %s\n\n", flagCacheInstances, dir)
+
+	clients := make([]posthog.Client, 0, flagCacheInstances)
+	caches := make([]*FileFlagCache, 0, flagCacheInstances)
+
+	for i := 0; i < flagCacheInstances; i++ {
+		cache, err := NewFileFlagCache(dir, "my-service-production")
+		if err != nil {
+			fmt.Printf("❌ Could not create the cache provider: %v\n", err)
+			return
+		}
+
+		client, err := posthog.NewWithConfig(projectAPIKey, posthog.Config{
+			Endpoint:                           endpoint,
+			SecretKey:                          secretKey,
+			DefaultFeatureFlagsPollingInterval: flagCachePollInterval,
+			FlagDefinitionCacheProvider:        cache,
+		})
+		if err != nil {
+			fmt.Printf("❌ Could not create the client: %v\n", err)
+			return
+		}
+
+		clients = append(clients, client)
+		caches = append(caches, cache)
+		fmt.Printf("   instance %s started\n", cache.shortID())
+	}
+
+	defer func() {
+		for i, client := range clients {
+			if err := client.Close(); err != nil {
+				fmt.Printf("   [%s] close: %v\n", caches[i].shortID(), err)
+			}
+		}
+	}()
+
+	time.Sleep(2 * flagCachePollInterval)
+
+	fmt.Println("\nEvaluating flags on every instance, entirely from the shared definitions:")
+	for i, client := range clients {
+		flags, err := client.GetAllFlags(posthog.FeatureFlagPayloadNoKey{
+			DistinctId:          "distinct-id-of-your-user",
+			OnlyEvaluateLocally: true,
+		})
+		if err != nil {
+			fmt.Printf("   [%s] %v\n", caches[i].shortID(), err)
+			continue
+		}
+		fmt.Printf("   [%s] evaluated %d flags locally\n", caches[i].shortID(), len(flags))
+	}
+}

examples/main.go

@@ -91,8 +91,9 @@ func showMenu() {
 	fmt.Println("4. Feature flag with SendFeatureFlagsOptions examples")
 	fmt.Println("5. Flag dependencies examples")
 	fmt.Println("6. ETag polling test (continuous, Ctrl+C to stop)")
-	fmt.Println("7. Run all examples (except ETag polling)")
-	fmt.Println("8. Exit")
+	fmt.Println("7. Distributed flag definition cache examples")
+	fmt.Println("8. Run all examples (except ETag polling)")
+	fmt.Println("9. Exit")
 }
 
 func runBasicCaptureExamples() {
@@ -125,6 +126,11 @@ func runETagPollingExample() {
 	TestETagPolling(projectAPIKey, secretKey, endpoint)
 }
 
+func runFlagDefinitionCacheExample() {
+	printExampleSection("DISTRIBUTED FLAG DEFINITION CACHE EXAMPLES")
+	TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint)
+}
+
 func printExampleSection(title string) {
 	fmt.Println("\n" + strings.Repeat("=", 60))
 	fmt.Println(title)
@@ -150,6 +156,9 @@ func runAllExamples() {
 
 	fmt.Printf("\n%s FLAG DEPENDENCIES %s\n", strings.Repeat("🔸", 20), strings.Repeat("🔸", 20))
 	TestFlagDependencies(projectAPIKey, secretKey, endpoint)
+
+	fmt.Printf("\n%s FLAG DEFINITION CACHE %s\n", strings.Repeat("🔸", 19), strings.Repeat("🔸", 19))
+	TestFlagDefinitionCache(projectAPIKey, secretKey, endpoint)
 }
 
 func isInteractive() bool {
@@ -170,7 +179,7 @@ func main() {
 
 	for {
 		showMenu()
-		choice := promptForInput("\nEnter your choice (1-8): ")
+		choice := promptForInput("\nEnter your choice (1-9): ")
 
 		switch choice {
 		case "1":
@@ -188,12 +197,14 @@ func main() {
 			// ETag polling runs continuously, so exit after it returns
 			return
 		case "7":
-			runAllExamples()
+			runFlagDefinitionCacheExample()
 		case "8":
+			runAllExamples()
+		case "9":
 			fmt.Println("👋 Goodbye!")
 			return
 		default:
-			fmt.Println("❌ Invalid choice. Please select 1-8.")
+			fmt.Println("❌ Invalid choice. Please select 1-9.")
 			continue
 		}