package alertmanager

import (
	"context"
	"flag"
	"time"

	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/pkg/errors"
	"github.com/prometheus/alertmanager/cluster/clusterpb"

	"github.com/cortexproject/cortex/pkg/alertmanager/alertspb"
	"github.com/cortexproject/cortex/pkg/alertmanager/alertstore"
	"github.com/cortexproject/cortex/pkg/util/services"
)

const (
	defaultPersistTimeout = 30 * time.Second
)

var (
	errInvalidPersistInterval = errors.New("invalid alertmanager persist interval, must be greater than zero")
)

type PersisterConfig struct {
	Interval time.Duration `yaml:"persist_interval"`
}

func (cfg *PersisterConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
	f.DurationVar(&cfg.Interval, prefix+".persist-interval", 15*time.Minute, "The interval between persisting the current alertmanager state (notification log and silences) to object storage. This is only used when sharding is enabled. This state is read when all replicas for a shard can not be contacted. In this scenario, having persisted the state more frequently will result in potentially fewer lost silences, and fewer duplicate notifications.")
}

func (cfg *PersisterConfig) Validate() error {
	if cfg.Interval <= 0 {
		return errInvalidPersistInterval
	}
	return nil
}

type PersistableState interface {
	State
	GetFullState() (*clusterpb.FullState, error)
}

// statePersister periodically writes the alertmanager state to persistent storage.
type statePersister struct {
	services.Service

	state  PersistableState
	store  alertstore.AlertStore
	userID string
	logger log.Logger

	timeout time.Duration
}

// newStatePersister creates a new state persister.
func newStatePersister(cfg PersisterConfig, userID string, state PersistableState, store alertstore.AlertStore, l log.Logger) *statePersister {

	s := &statePersister{
		state:   state,
		store:   store,
		userID:  userID,
		logger:  l,
		timeout: defaultPersistTimeout,
	}

	s.Service = services.NewTimerService(cfg.Interval, s.starting, s.iteration, nil)

	return s
}

func (s *statePersister) starting(ctx context.Context) error {
	// Waits until the state replicator is settled, so that state is not
	// persisted before obtaining some initial state.
	return s.state.WaitReady(ctx)
}

func (s *statePersister) iteration(ctx context.Context) error {
	if err := s.persist(ctx); err != nil {
		level.Error(s.logger).Log("msg", "failed to persist state", "user", s.userID, "err", err)
	}
	return nil
}

func (s *statePersister) persist(ctx context.Context) error {
	// Only the replica at position zero should write the state.
	if s.state.Position() != 0 {
		return nil
	}

	level.Debug(s.logger).Log("msg", "persisting state", "user", s.userID)

	fs, err := s.state.GetFullState()
	if err != nil {
		return err
	}

	ctx, cancel := context.WithTimeout(ctx, s.timeout)
	defer cancel()

	desc := alertspb.FullStateDesc{State: fs}
	if err := s.store.SetFullState(ctx, s.userID, desc); err != nil {
		return err
	}

	return nil
}
