tidb/timer/api/client.go

// Copyright 2023 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package api

import (
	"context"
	"strings"
	"time"
	"unsafe"

	"github.com/pingcap/errors"
)

// GetTimerOption is the option to get timers
type GetTimerOption func(*TimerCond)

// WithKey indicates to get a timer with the specified key
func WithKey(key string) GetTimerOption {
	return func(cond *TimerCond) {
		cond.Key.Set(key)
		cond.KeyPrefix = false
	}
}

// WithKeyPrefix to get timers with the indicated key prefix
func WithKeyPrefix(keyPrefix string) GetTimerOption {
	return func(cond *TimerCond) {
		cond.Key.Set(keyPrefix)
		cond.KeyPrefix = true
	}
}

// WithID indicates to get a timer with the specified id
func WithID(id string) GetTimerOption {
	return func(cond *TimerCond) {
		cond.ID.Set(id)
	}
}

// UpdateTimerOption is the option to update the timer
type UpdateTimerOption func(*TimerUpdate)

// WithSetEnable indicates to set the timer's `Enable` field
func WithSetEnable(enable bool) UpdateTimerOption {
	return func(update *TimerUpdate) {
		update.Enable.Set(enable)
	}
}

// WithSetSchedExpr indicates to set the timer's schedule policy
func WithSetSchedExpr(tp SchedPolicyType, expr string) UpdateTimerOption {
	return func(update *TimerUpdate) {
		update.SchedPolicyType.Set(tp)
		update.SchedPolicyExpr.Set(expr)
	}
}

// WithSetWatermark indicates to set the timer's watermark
func WithSetWatermark(watermark time.Time) UpdateTimerOption {
	return func(update *TimerUpdate) {
		update.Watermark.Set(watermark)
	}
}

// WithSetSummaryData indicates to set the timer's summary
func WithSetSummaryData(summary []byte) UpdateTimerOption {
	return func(update *TimerUpdate) {
		update.SummaryData.Set(summary)
	}
}

// TimerClient is an interface exposed to user to manage timers
type TimerClient interface {
	// GetDefaultNamespace returns the default namespace of this client
	GetDefaultNamespace() string
	// CreateTimer creates a new timer
	CreateTimer(ctx context.Context, spec TimerSpec) (*TimerRecord, error)
	// GetTimerByID queries the timer by ID
	GetTimerByID(ctx context.Context, timerID string) (*TimerRecord, error)
	// GetTimerByKey queries the timer by key
	GetTimerByKey(ctx context.Context, key string) (*TimerRecord, error)
	// GetTimers queries timers by options
	GetTimers(ctx context.Context, opts ...GetTimerOption) ([]*TimerRecord, error)
	// UpdateTimer updates a timer
	UpdateTimer(ctx context.Context, timerID string, opts ...UpdateTimerOption) error
	// CloseTimerEvent closes the triggering event of a timer
	CloseTimerEvent(ctx context.Context, timerID string, eventID string, opts ...UpdateTimerOption) error
	// DeleteTimer deletes a timer
	DeleteTimer(ctx context.Context, timerID string) (bool, error)
}

// DefaultStoreNamespace is the default namespace
const DefaultStoreNamespace = "default"

// defaultTimerClient is the default implement of timer client
type defaultTimerClient struct {
	namespace string
	store     *TimerStore
}

// NewDefaultTimerClient creates a new defaultTimerClient
func NewDefaultTimerClient(store *TimerStore) TimerClient {
	return &defaultTimerClient{
		namespace: DefaultStoreNamespace,
		store:     store,
	}
}

func (c *defaultTimerClient) GetDefaultNamespace() string {
	return c.namespace
}

func (c *defaultTimerClient) CreateTimer(ctx context.Context, spec TimerSpec) (*TimerRecord, error) {
	if spec.Namespace == "" {
		spec.Namespace = c.namespace
	}

	timerID, err := c.store.Create(ctx, &TimerRecord{
		TimerSpec: spec,
	})

	if err != nil {
		return nil, err
	}
	return c.store.GetByID(ctx, timerID)
}

func (c *defaultTimerClient) GetTimerByID(ctx context.Context, timerID string) (*TimerRecord, error) {
	return c.store.GetByID(ctx, timerID)
}

func (c *defaultTimerClient) GetTimerByKey(ctx context.Context, key string) (*TimerRecord, error) {
	return c.store.GetByKey(ctx, c.namespace, key)
}

func (c *defaultTimerClient) GetTimers(ctx context.Context, opts ...GetTimerOption) ([]*TimerRecord, error) {
	cond := &TimerCond{}
	for _, opt := range opts {
		opt(cond)
	}
	return c.store.List(ctx, cond)
}

func (c *defaultTimerClient) UpdateTimer(ctx context.Context, timerID string, opts ...UpdateTimerOption) error {
	update := &TimerUpdate{}
	for _, opt := range opts {
		opt(update)
	}
	return c.store.Update(ctx, timerID, update)
}

func (c *defaultTimerClient) CloseTimerEvent(ctx context.Context, timerID string, eventID string, opts ...UpdateTimerOption) error {
	update := &TimerUpdate{}
	for _, opt := range opts {
		opt(update)
	}

	fields := update.FieldsSet(unsafe.Pointer(&update.Watermark), unsafe.Pointer(&update.SummaryData))
	if len(fields) > 0 {
		return errors.Errorf("The field(s) [%s] are not allowed to update when close event", strings.Join(fields, ", "))
	}

	timer, err := c.GetTimerByID(ctx, timerID)
	if err != nil {
		return err
	}

	var zeroTime time.Time
	update.CheckEventID.Set(eventID)
	update.EventStatus.Set(SchedEventIdle)
	update.EventID.Set("")
	update.EventData.Set(nil)
	update.EventStart.Set(zeroTime)
	if !update.Watermark.Present() {
		update.Watermark.Set(timer.EventStart)
	}
	return c.store.Update(ctx, timerID, update)
}

func (c *defaultTimerClient) DeleteTimer(ctx context.Context, timerID string) (bool, error) {
	return c.store.Delete(ctx, timerID)
}