upstream-mattermost/server/channels/app/channel_guards.go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.

package app

import (
	"net/http"
	"sync"
	"time"

	"github.com/mattermost/mattermost/server/public/model"
	"github.com/mattermost/mattermost/server/public/shared/mlog"
	"github.com/mattermost/mattermost/server/public/shared/request"
	"github.com/mattermost/mattermost/server/v8/channels/store"
)

// Backoff bounds for the guard-cache reload retry. Package vars (not consts) so tests can shrink
// them via t.Cleanup-restored override.
var (
	guardCacheRetryInitialDelay = 1 * time.Second
	guardCacheRetryMaxDelay     = 5 * time.Minute
)

const clusterEventInvalidateChannelGuardCache = model.ClusterEvent("inv_channel_guards")

// reloadGuardCache scans the ChannelGuards table and atomically replaces the in-memory cache with
// the result. Used both at startup (from NewChannels) and from the cluster invalidation handler.
// Forces a master read because all callers (post-write reload, cluster invalidation) can race with
// replica lag.
func (ch *Channels) reloadGuardCache(rctx request.CTX, s store.Store) error {
	guards, err := s.ChannelGuard().GetAll(store.RequestContextWithMaster(rctx))
	if err != nil {
		return err
	}

	fresh := &sync.Map{}
	grouped := map[string][]*store.ChannelGuard{}
	for _, g := range guards {
		grouped[g.ChannelId] = append(grouped[g.ChannelId], g)
	}
	for channelID, slice := range grouped {
		fresh.Store(channelID, slice)
	}

	ch.guardCache.Store(fresh)
	return nil
}

// getGuardsForChannel returns the cached guard slice for a channel, or nil if none.
func (ch *Channels) getGuardsForChannel(channelID string) []*store.ChannelGuard {
	m := ch.guardCache.Load()
	if m == nil {
		return nil
	}
	v, ok := m.Load(channelID)
	if !ok {
		return nil
	}
	guards, _ := v.([]*store.ChannelGuard)
	return guards
}

// clusterInvalidateGuardCacheHandler is registered as the receive-side handler for
// clusterEventInvalidateChannelGuardCache. The handler refetches the entire table.
func (ch *Channels) clusterInvalidateGuardCacheHandler(msg *model.ClusterMessage) {
	rctx := request.EmptyContext(ch.srv.Log())
	if err := ch.reloadGuardCache(rctx, ch.srv.Store()); err != nil {
		ch.srv.Log().Warn(
			"Failed to reload channel guard cache after cluster invalidation; retry scheduled",
			mlog.String("event", string(msg.Event)),
			mlog.Err(err),
		)
		ch.scheduleGuardCacheReloadRetry()
	}
}

// broadcastChannelGuardInvalidation tells the rest of the cluster to refetch their guard caches.
// The payload is intentionally empty.
func (ch *Channels) broadcastChannelGuardInvalidation() {
	cluster := ch.srv.platform.Cluster()
	if cluster == nil {
		return
	}

	msg := &model.ClusterMessage{
		Event:            clusterEventInvalidateChannelGuardCache,
		SendType:         model.ClusterSendReliable,
		WaitForAllToSend: true,
	}
	cluster.SendClusterMessage(msg)
}

// RegisterChannelGuard records that pluginID claims channelID. The caller's pluginID is expected to
// be lowercased.
func (a *App) RegisterChannelGuard(rctx request.CTX, channelID, pluginID string) *model.AppError {
	if channelID == "" {
		return model.NewAppError("RegisterChannelGuard", "app.channel_guard.register.empty_channel.app_error", nil, "", http.StatusBadRequest)
	}
	if !model.IsValidId(channelID) {
		return model.NewAppError("RegisterChannelGuard", "app.channel_guard.invalid_channel.app_error", nil, "", http.StatusBadRequest)
	}

	guard := &store.ChannelGuard{
		ChannelId: channelID,
		PluginId:  pluginID,
		CreatedAt: model.GetMillis(),
	}
	if err := a.Srv().Store().ChannelGuard().Save(rctx, guard); err != nil {
		return model.NewAppError("RegisterChannelGuard", "app.channel_guard.register.app_error", nil, err.Error(), http.StatusInternalServerError).Wrap(err)
	}

	ch := a.Channels()
	if err := ch.reloadGuardCache(rctx, a.Srv().Store()); err != nil {
		a.Srv().Log().Warn(
			"Failed to reload channel guard cache after Register; retry scheduled",
			mlog.String("channel_id", channelID),
			mlog.String("plugin_id", pluginID),
			mlog.Err(err),
		)
		ch.scheduleGuardCacheReloadRetry()
	}
	ch.broadcastChannelGuardInvalidation()
	return nil
}

// UnregisterChannelGuard removes pluginID's claim on channelID. If pluginID has no claim on the
// channel, this is a no-op (returns nil). The store-level DELETE matches by both ChannelId and
// PluginId, so other plugins' claims on the same channel are left untouched.
func (a *App) UnregisterChannelGuard(rctx request.CTX, channelID, pluginID string) *model.AppError {
	if channelID == "" {
		return model.NewAppError("UnregisterChannelGuard", "app.channel_guard.unregister.empty_channel.app_error", nil, "", http.StatusBadRequest)
	}
	if !model.IsValidId(channelID) {
		return model.NewAppError("UnregisterChannelGuard", "app.channel_guard.invalid_channel.app_error", nil, "", http.StatusBadRequest)
	}

	rowsAffected, err := a.Srv().Store().ChannelGuard().Delete(rctx, channelID, pluginID)
	if err != nil {
		return model.NewAppError("UnregisterChannelGuard", "app.channel_guard.unregister.app_error", nil, err.Error(), http.StatusInternalServerError).Wrap(err)
	}
	if rowsAffected == 0 {
		a.Srv().Log().Warn(
			"UnregisterChannelGuard removed no rows; pluginID does not match any guard for this channel",
			mlog.String("error_id", "unregister_no_matching_guard"),
			mlog.String("channel_id", channelID),
			mlog.String("plugin_id", pluginID),
		)
	}

	ch := a.Channels()
	if err := ch.reloadGuardCache(rctx, a.Srv().Store()); err != nil {
		a.Srv().Log().Warn(
			"Failed to reload channel guard cache after Unregister; retry scheduled",
			mlog.String("channel_id", channelID),
			mlog.String("plugin_id", pluginID),
			mlog.Err(err),
		)
		ch.scheduleGuardCacheReloadRetry()
	}
	ch.broadcastChannelGuardInvalidation()
	return nil
}

// scheduleGuardCacheReloadRetry kicks off a single in-flight retry goroutine that calls
// reloadGuardCache with exponential backoff until success or until the server is shutting down.
// Multiple concurrent calls collapse to a single retry — useful when Register, Unregister, the
// cluster handler, and the startup loader can all see the same DB outage simultaneously.
//
// Returns true if a new retry goroutine was scheduled, false if one was already in flight. Call
// sites can ignore the return value; tests use it to assert single-flight semantics.
func (ch *Channels) scheduleGuardCacheReloadRetry() bool {
	if !ch.guardCacheRetryInFlight.CompareAndSwap(false, true) {
		return false
	}
	go ch.runGuardCacheReloadRetry()
	return true
}

func (ch *Channels) runGuardCacheReloadRetry() {
	defer ch.guardCacheRetryInFlight.Store(false)
	rctx := request.EmptyContext(ch.srv.Log())

	delay := guardCacheRetryInitialDelay
	for attempt := 1; ; attempt++ {
		timer := time.NewTimer(delay)
		select {
		case <-ch.interruptQuitChan:
			timer.Stop()
			ch.srv.Log().Info(
				"Channel guard cache reload retry cancelled by shutdown",
				mlog.Int("attempt", attempt),
			)
			return
		case <-timer.C:
		}

		if err := ch.reloadGuardCache(rctx, ch.srv.Store()); err != nil {
			ch.srv.Log().Info(
				"Channel guard cache reload retry attempt failed; will retry",
				mlog.Int("attempt", attempt),
				mlog.Err(err),
			)
			delay *= 2
			if delay > guardCacheRetryMaxDelay {
				delay = guardCacheRetryMaxDelay
			}
			continue
		}

		ch.srv.Log().Info(
			"Channel guard cache reload retry succeeded",
			mlog.Int("attempt", attempt),
		)
		return
	}
}