more cart

pkg/profile/email_index.go

@@ -0,0 +1,126 @@
+package profile
+
+import (
+	"bytes"
+	"io"
+	"context"
+	"log"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"git.k6n.net/mats/go-cart-actor/pkg/actor"
+	"google.golang.org/protobuf/proto"
+)
+
+// emailIndexEntry captures the email preferences snapshot for a profile.
+type emailIndexEntry struct {
+	ID               uint64
+	EmailPreferences *EmailPreferences
+}
+
+// ProfileEmailIndex is a thread-safe in-memory map from email address to
+// profile ID and email preferences snapshot. It is populated at startup by
+// scanning the profile storage directory, and updated when SetProfile
+// mutations change the email or preferences.
+type ProfileEmailIndex struct {
+	mu   sync.RWMutex
+	byID map[uint64]string          // profileID → email (for cleanup on email change)
+	idx  map[string]emailIndexEntry // email → profile info
+}
+
+// NewProfileEmailIndex returns an empty index.
+func NewProfileEmailIndex() *ProfileEmailIndex {
+	return &ProfileEmailIndex{
+		byID: make(map[uint64]string),
+		idx:  make(map[string]emailIndexEntry),
+	}
+}
+
+// Set records or updates the email→profile mapping. When email is empty the
+// entry is removed (profile was anonymized).
+func (ix *ProfileEmailIndex) Set(profileID uint64, email string, prefs *EmailPreferences) {
+	ix.mu.Lock()
+	defer ix.mu.Unlock()
+
+	// Remove any previous email for this profile.
+	if oldEmail, ok := ix.byID[profileID]; ok {
+		delete(ix.idx, oldEmail)
+	}
+
+	if strings.TrimSpace(email) == "" {
+		delete(ix.byID, profileID)
+		return
+	}
+
+	email = strings.ToLower(strings.TrimSpace(email))
+	ix.byID[profileID] = email
+	ix.idx[email] = emailIndexEntry{
+		ID:               profileID,
+		EmailPreferences: prefs,
+	}
+}
+
+// Lookup returns the email preferences for the given email. The second return
+// value is false when the email is not found.
+func (ix *ProfileEmailIndex) Lookup(email string) (emailIndexEntry, bool) {
+	ix.mu.RLock()
+	defer ix.mu.RUnlock()
+	entry, ok := ix.idx[strings.ToLower(strings.TrimSpace(email))]
+	return entry, ok
+}
+
+// BuildEmailIndex populates the index by scanning the profile storage
+// directory and replaying each event log into a fresh ProfileGrain. It must
+// be called once at service startup after the mutation registry and storage
+// are initialised.
+//
+// dir  — the same path passed to actor.NewDiskStorage (e.g. "data/profiles").
+// replayer — an *actor.StateStorage for reading event log files.
+// reg  — the profile mutation registry.
+// ix   — the (empty) index to populate.
+func BuildEmailIndex(dir string, replayer interface {
+	Load(r io.Reader, onMessage func(msg proto.Message, timeStamp time.Time)) error
+}, reg actor.MutationRegistry, ix *ProfileEmailIndex) {
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		log.Printf("email index: cannot read profile dir %s: %v — starting with empty index", dir, err)
+		return
+	}
+
+	var count int
+	ctx := context.Background()
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".events.log") {
+			continue
+		}
+		idStr := strings.TrimSuffix(entry.Name(), ".events.log")
+		id, err := strconv.ParseUint(idStr, 10, 64)
+		if err != nil {
+			continue
+		}
+
+		p := NewProfileGrain(id, time.Now())
+		data, err := os.ReadFile(filepath.Join(dir, entry.Name()))
+		if err != nil {
+			log.Printf("email index: read %s: %v", entry.Name(), err)
+			continue
+		}
+		if err := replayer.Load(bytes.NewReader(data), func(msg proto.Message, _ time.Time) { //nolint:govet
+			_, _ = reg.Apply(ctx, p, msg)
+		}); err != nil {
+			log.Printf("email index: replay %s: %v", entry.Name(), err)
+			continue
+		}
+
+		if p.Email != "" {
+			ix.Set(p.Id, p.Email, p.EmailPreferences)
+			count++
+		}
+	}
+
+	log.Printf("email index: built %d entries from %s", count, dir)
+}