API: Adds DeletePrefixes (#37)

Adds a new DeletePrefixes API which deletes all jobs (as well as
counters) whose names match the given prefixes. Note the given strings
to this API are prefixes of the _job names_, not the underlying ETCD
keys. Prefixes will be serialized into the actual ETCD keys.

This is useful for Dapr to add a feature whereby jobs (of all of types)
in a particular namespace (some job prefix) are deleted when a
Kubernetes Namespace is deleted.

Moves `/internal/tests` to `/tests` to expose cron ETCD helper funcs for
library consumers. Expands package with a `cron` sub-package to introduce
single and clustered cron server setup helpers.

Exposes the `CounterGarbageCollectionInterval` time duration pointer
option to enable testing time interval triggered counter garbage
collection. This is set to a low value during some tests, which has a
side effect of soft testing this doesn't prematurely delete counters.

Slight refactor of `api_test.go` to reside in a different package to
prevent a circular import from `/tests`.

Adds `/fake` for exposing a fake cron that library consumers can use for
testing.

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

api.go

@@ -9,10 +9,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strings"
 
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"google.golang.org/protobuf/proto"
-	"k8s.io/apimachinery/pkg/util/validation"
 
 	"github.com/diagridio/go-etcd-cron/api"
 )
@@ -103,8 +102,6 @@ func (c *cron) Delete(ctx context.Context, name string) error {
 	if _, err := c.client.Delete(ctx, jobKey); err != nil {
 		return err
 	}
-	counterKey := c.key.CounterKey(name)
-	c.collector.Push(counterKey)
 
 	if _, ok := c.queueCache.Load(jobKey); !ok {
 		return nil
@@ -114,17 +111,54 @@ func (c *cron) Delete(ctx context.Context, name string) error {
 	return c.queue.Dequeue(jobKey)
 }
 
-// validateName validates the name of a job.
-func (c *cron) validateName(name string) error {
-	if len(name) == 0 {
-		return errors.New("job name cannot be empty")
+// DeletePrefixes deletes cron jobs with the given prefixes from the cron
+// instance.
+func (c *cron) DeletePrefixes(ctx context.Context, prefixes ...string) error {
+	select {
+	case <-c.readyCh:
+	case <-c.closeCh:
+		return errors.New("cron is closed")
+	case <-ctx.Done():
+		return ctx.Err()
 	}
 
-	for _, segment := range strings.Split(strings.ToLower(c.validateNameReplacer.Replace(name)), "||") {
-		if errs := validation.IsDNS1123Subdomain(segment); len(errs) > 0 {
-			return fmt.Errorf("job name is invalid %q: %s", name, strings.Join(errs, ", "))
+	for _, prefix := range prefixes {
+		if len(prefix) == 0 {
+			continue
+		}
+
+		if err := c.validateName(prefix); err != nil {
+			return err
+		}
+	}
+
+	var errs []error
+	removeFromCache := func(jobKey string) {
+		c.queueLock.Lock(jobKey)
+		defer c.queueLock.DeleteUnlock(jobKey)
+
+		if _, ok := c.queueCache.Load(jobKey); ok {
+			return
+		}
+
+		c.queueCache.Delete(jobKey)
+		errs = append(errs, c.queue.Dequeue(jobKey))
+	}
+
+	for _, prefix := range prefixes {
+		keyPrefix := c.key.JobKey(prefix)
+		c.log.V(3).Info("deleting jobs with prefix", "prefix", keyPrefix)
+
+		resp, err := c.client.Delete(ctx, keyPrefix, clientv3.WithPrefix(), clientv3.WithPrevKV())
+		if err != nil {
+			errs = append(errs, fmt.Errorf("failed to delete jobs with prefix %q: %w", prefix, err))
+			continue
+		}
+
+		for _, kv := range resp.PrevKvs {
+			removeFromCache(string(kv.Key))
 		}
 	}
 
-	return nil
+	return errors.Join(errs...)
 }