internal/store/postgres/channel_dialog_cache.go

Rama main · réplica de lectura del repositorio original.

package postgres

import (
	"context"
	"sync"

	"telesrv/internal/domain"
	"telesrv/internal/readmodelcache"
)

type channelDialogCacheKey struct {
	userID    int64
	channelID int64
}

type channelDialogCacheEntry struct {
	dialog             domain.ChannelDialog
	listVisible        bool
	topMentioned       bool
	topMediaUnread     bool
	topUnreadProjected bool
}

// ChannelDialogCache caches the viewer-scoped channel dialog projection, backed by the common cache
// primitive readmodelcache.Cache (single-entry LRU eviction / epoch guard / built-in singleflight).
//
// Used only on pool (non-transactional) read paths. The cached value depends on three read models:
// channel_base, channel_member(viewer,channel), and dialog_light(viewer,channel); ReadModelChangeListener
// invalidates the corresponding keys on write-side NOTIFY and flushes on reconnect. warm-from-list backfills via put (including DefaultSendAs).
type ChannelDialogCache struct {
	cache *readmodelcache.Cache[channelDialogCacheKey, channelDialogCacheEntry]

	indexMu     sync.Mutex
	channelKeys map[int64]map[channelDialogCacheKey]struct{}
}

func NewChannelDialogCache(max int) *ChannelDialogCache {
	c := &ChannelDialogCache{channelKeys: make(map[int64]map[channelDialogCacheKey]struct{})}
	cache := readmodelcache.New[channelDialogCacheKey, channelDialogCacheEntry](readmodelcache.Config[channelDialogCacheKey, channelDialogCacheEntry]{
		MaxEntries: max,
		Clone:      cloneChannelDialogCacheEntry,
		OnStore:    c.indexEntry,
		OnRemove:   c.unindexEntry,
	})
	if cache == nil {
		return nil
	}
	c.cache = cache
	return c
}

func (c *ChannelDialogCache) get(userID, channelID int64) (domain.ChannelDialog, bool) {
	if c == nil || userID == 0 || channelID == 0 {
		return domain.ChannelDialog{}, false
	}
	entry, ok := c.cache.Peek(channelDialogCacheKey{userID: userID, channelID: channelID})
	return entry.dialog, ok
}

func (c *ChannelDialogCache) getListProjection(userID, channelID int64) (channelDialogCacheEntry, bool) {
	if c == nil || userID == 0 || channelID == 0 {
		return channelDialogCacheEntry{}, false
	}
	entry, ok := c.cache.Peek(channelDialogCacheKey{userID: userID, channelID: channelID})
	return entry, ok && entry.listVisible
}

func (c *ChannelDialogCache) getOrLoad(ctx context.Context, userID, channelID int64, load func() (domain.ChannelDialog, error)) (domain.ChannelDialog, error) {
	if c == nil || userID == 0 || channelID == 0 {
		return load()
	}
	entry, err := c.cache.GetOrLoad(ctx, channelDialogCacheKey{userID: userID, channelID: channelID}, func() (channelDialogCacheEntry, error) {
		dialog, err := load()
		return channelDialogCacheEntry{dialog: dialog}, err
	})
	return entry.dialog, err
}

func (c *ChannelDialogCache) put(dialog domain.ChannelDialog) {
	if c == nil || dialog.UserID == 0 || dialog.ChannelID == 0 {
		return
	}
	c.cache.Store(channelDialogCacheKey{userID: dialog.UserID, channelID: dialog.ChannelID}, channelDialogCacheEntry{dialog: dialog})
}

// cacheEpoch snapshots the epoch before a "list warm write-back"; together with putIfEpoch it closes the
// warm-vs-invalidation race: if an invalidation arrives (epoch increments) between the DB query returning and the write-back, the stale projection write-back is rejected.
func (c *ChannelDialogCache) cacheEpoch() uint64 {
	if c == nil {
		return 0
	}
	return c.cache.LoadEpoch()
}

func (c *ChannelDialogCache) putIfEpoch(dialog domain.ChannelDialog, loadEpoch uint64) {
	if c == nil || dialog.UserID == 0 || dialog.ChannelID == 0 {
		return
	}
	c.cache.StoreIfEpoch(channelDialogCacheKey{userID: dialog.UserID, channelID: dialog.ChannelID}, channelDialogCacheEntry{dialog: dialog}, loadEpoch)
}

func (c *ChannelDialogCache) putListProjectionIfEpoch(
	dialog domain.ChannelDialog,
	topMentioned bool,
	topMediaUnread bool,
	topUnreadProjected bool,
	loadEpoch uint64,
) {
	if c == nil || dialog.UserID == 0 || dialog.ChannelID == 0 {
		return
	}
	c.cache.StoreIfEpoch(
		channelDialogCacheKey{userID: dialog.UserID, channelID: dialog.ChannelID},
		channelDialogCacheEntry{
			dialog:             dialog,
			listVisible:        true,
			topMentioned:       topMentioned,
			topMediaUnread:     topMediaUnread,
			topUnreadProjected: topUnreadProjected,
		},
		loadEpoch,
	)
}

func (c *ChannelDialogCache) delete(userID, channelID int64) {
	if c == nil || userID == 0 || channelID == 0 {
		return
	}
	c.cache.Invalidate(channelDialogCacheKey{userID: userID, channelID: channelID})
}

func (c *ChannelDialogCache) deleteChannel(channelID int64) {
	if c == nil || channelID == 0 {
		return
	}
	c.indexMu.Lock()
	indexed := c.channelKeys[channelID]
	keys := make([]channelDialogCacheKey, 0, len(indexed))
	for key := range indexed {
		keys = append(keys, key)
	}
	c.indexMu.Unlock()
	c.cache.Invalidate(keys...)
}

func (c *ChannelDialogCache) flush() {
	if c == nil {
		return
	}
	c.cache.Flush()
	c.indexMu.Lock()
	c.channelKeys = make(map[int64]map[channelDialogCacheKey]struct{})
	c.indexMu.Unlock()
}

func (c *ChannelDialogCache) indexEntry(key channelDialogCacheKey, _ channelDialogCacheEntry) {
	c.indexMu.Lock()
	keys := c.channelKeys[key.channelID]
	if keys == nil {
		keys = make(map[channelDialogCacheKey]struct{})
		c.channelKeys[key.channelID] = keys
	}
	keys[key] = struct{}{}
	c.indexMu.Unlock()
}

func (c *ChannelDialogCache) unindexEntry(key channelDialogCacheKey, _ channelDialogCacheEntry) {
	c.indexMu.Lock()
	keys := c.channelKeys[key.channelID]
	delete(keys, key)
	if len(keys) == 0 {
		delete(c.channelKeys, key.channelID)
	}
	c.indexMu.Unlock()
}

func cloneChannelDialogCacheEntry(entry channelDialogCacheEntry) channelDialogCacheEntry {
	if entry.dialog.DefaultSendAs != nil {
		peer := *entry.dialog.DefaultSendAs
		entry.dialog.DefaultSendAs = &peer
	}
	return entry
}