func (l *dbLimiter) deleteOldUsageData(ctx context.Context) error {
	deleteEpoch := l.getTime().Add(-defaultDeleteUsageOlderThan).Truncate(l.defaultWriterLimit.Period).Unix()
	res, err := l.db.ExecContext(ctx, "delete from writer_usage where bucket < ?", deleteEpoch)
	if err != nil {
		return errors.Wrap(err, "could not delete from writer_usage table")
	}
	rows, err := res.RowsAffected()
	if err != nil {
		return errors.Wrap(err, "could not get rows affected after deleting from writer_usage")
	}
	if rows > 0 {
		events.Log("deleted %{rows}d rows from the writer_usage table", rows)
	}
	stats.Add("writer-usage-rows-deleted", rows)
	return nil
}
deleteOldUsageData cleans the writer_usage table of old entries
package executive
import (
	"context"
	"database/sql"
	"sync"
	"time"
	"github.com/pkg/errors"
	"github.com/segmentio/ctlstore/pkg/errs"
	"github.com/segmentio/ctlstore/pkg/limits"
	"github.com/segmentio/ctlstore/pkg/schema"
	"github.com/segmentio/ctlstore/pkg/utils"
	"github.com/segmentio/events/v2"
	"github.com/segmentio/stats/v4"
)
const (
	defaultRefreshPeriod        = 1 * time.Minute  
	defaultDeleteUsagePeriod    = 1 * time.Hour    
	defaultDeleteUsageOlderThan = 24 * time.Hour   
)
type (
	dbLimiter struct {
		db                 *sql.DB
		tableSizer         *tableSizer
		mut                sync.Mutex  
		defaultWriterLimit limits.RateLimit
		perWriterLimits    map[string]int64  
		timeFunc           func() time.Time
	}
	limiterRequest struct {
		writerName string
		familyName string
		requests   []ExecutiveMutationRequest
	}
)
func newDBLimiter(db *sql.DB, dbType string, defaultTableLimit limits.SizeLimits, writerLimitPeriod time.Duration, writerLimit int64) *dbLimiter {
	return &dbLimiter{
		db:                 db,
		tableSizer:         newTableSizer(db, dbType, defaultTableLimit, time.Minute),
		defaultWriterLimit: limits.RateLimit{Amount: writerLimit, Period: writerLimitPeriod},
		perWriterLimits:    make(map[string]int64),
	}
}
func (l *dbLimiter) allowed(ctx context.Context, tx *sql.Tx, lr limiterRequest) (bool, error) {
	if err := l.checkTableSizes(ctx, lr); err != nil {
		return false, errors.Wrap(err, "check table sizes")
	}
	allowed, err := l.checkWriterRates(ctx, tx, lr)
	if err != nil {
		return false, errors.Wrap(err, "check writer rates")
	}
	return allowed, nil
}
func (l *dbLimiter) checkWriterRates(ctx context.Context, tx *sql.Tx, lr limiterRequest) (bool, error) {
	numMutations := len(lr.requests)
	if numMutations == 0 {
		return true, nil
	}
	bucket := l.periodEpoch()
	row := tx.QueryRowContext(ctx, "SELECT amount FROM writer_usage WHERE writer_name=? AND bucket=?", lr.writerName, bucket)
	var amount int64
	err := row.Scan(&amount)
	if err != nil && err != sql.ErrNoRows {
		return false, errors.Wrap(err, "select from writer_usage")
	}
	amount += int64(numMutations)
	if err == sql.ErrNoRows {
		res, err := tx.ExecContext(ctx, "INSERT INTO writer_usage (bucket,writer_name,amount) VALUES (?,?,?)",
			bucket, lr.writerName, amount)
		if err != nil {
			return false, errors.Wrap(err, "insert into writer_usage")
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return false, errors.Wrap(err, "affected rows from insert into writer_usage")
		}
		if rowsAffected == 0 {
			return false, errors.New("insert into writer_usage failed (no rows updated)")
		}
	} else {
		res, err := tx.ExecContext(ctx, "UPDATE writer_usage SET amount=? where bucket=? and writer_name=?",
			amount, bucket, lr.writerName)
		if err != nil {
			return false, errors.Wrap(err, "update writer_usage")
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return false, errors.Wrap(err, "affected rows from update writer_usage")
		}
		if rowsAffected == 0 {
			return false, errors.New("updating writer_usage failed (no rows updated)")
		}
	}
	writerLimit := l.limitForWriter(lr.writerName)
	allowed := amount <= writerLimit
	events.Debug("limiter: writer:%v writerLimit:%v amount:%v allowed:%v", lr.writerName, writerLimit, amount, allowed)
	return allowed, nil
}
func (l *dbLimiter) checkTableSizes(ctx context.Context, lr limiterRequest) error {
	tables := make(map[schema.FamilyTable]struct{})
	for _, req := range lr.requests {
		ft := schema.FamilyTable{Family: lr.familyName, Table: req.TableName}
		if _, ok := tables[ft]; !ok {
			tables[ft] = struct{}{}  
			if _, err := l.tableSizer.tableOK(ft); err != nil {
				return err
			}
		}
	}
	return nil
}
func (l *dbLimiter) start(ctx context.Context) error {
	events.Log("Starting the db limiter")
	if err := l.tableSizer.start(ctx); err != nil {
		return errors.Wrap(err, "could not start sizer")
	}
	instrumentUpdateErr := func(err error) {
		errs.IncrDefault(stats.Tag{Name: "op", Value: "update-limits"})
	}
	if err := l.refreshWriterLimits(ctx); err != nil {
		instrumentUpdateErr(err)
		return errors.Wrap(err, "refresh writer limits")
	}
	go utils.CtxLoop(ctx, defaultRefreshPeriod, func() {
		if err := l.refreshWriterLimits(ctx); err != nil {
			events.Log("could not update limits: %{error}s", err)
			instrumentUpdateErr(err)
		}
	})
	go utils.CtxLoop(ctx, defaultDeleteUsagePeriod, func() {
		if err := l.deleteOldUsageData(ctx); err != nil {
			events.Log("could not collect garbage %{error}s", err)
			errs.IncrDefault(stats.Tag{Name: "op", Value: "limiter-collect-garbage"})
		}
	})
	return nil
}