List API (#43)

* List API

Adds a new `List` API to the cron service. This API returns the list of
current committed active jobs that have a name matching the given
prefix. An empty prefix will return all jobs.

```go
List(ctx context.Context, prefix string) (*ListResponse, error)
```

A new `ListResponse` proto message has been added which contains a list
of named Jobs.

As part of this change to clean up the codebase, the cron API definition
has been moved into `api/api.go`. A new `queue` runnable has been
created which is responsible for managing the underlying job queue based
on events from the informer. A new `internal/api` package is responsible
for implementing the server CRUD APIs.

Uses dapr/kit concurrency/map from dapr/kit#104

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

* Skip cache

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

* Decrease number of jobs to delete

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

* Updates dapr/kit to origin remote HEAD

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

* Adds comments around `List` `more` usage and requested limits.

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

* Removes `more` from `ListResponse` proto message as it is not used

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

* Fix defer triggered

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

* Updates gomod2nix.yaml

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

---------

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

Author: JoshVanL

.github/workflows/test.yaml

@@ -11,4 +11,5 @@ jobs:
     - uses: golangci/[email protected]
       with:
         version: v1.61.0
+        skip-cache: true
     - run: go test -v --race ./...

Makefile

@@ -13,7 +13,7 @@ endif
 
 .PHONY: lint
 lint:
-	$(GOLANGCI_LINT) run --timeout=20m
+	$(GOLANGCI_LINT) run --timeout=20m ./...
 
 
 ################################################################################
@@ -42,7 +42,6 @@ PROTOC_GEN_GO_VERSION = v1.32.0
 
 PROTOC_GEN_GO_GRPC_VERSION = 1.3.0
 
-PROTOS:=$(shell ls proto)
 PROTO_PREFIX:=github.com/diagridio/go-etcd-cron
 
 .PHONY: check-proto-version
@@ -56,20 +55,16 @@ check-proto-version: ## Checking the version of proto related tools
 	@test "$(shell protoc-gen-go --version 2>&1)" = "protoc-gen-go $(PROTOC_GEN_GO_VERSION)" \
 	|| { echo "please use protoc-gen-go $(PROTOC_GEN_GO_VERSION) to generate proto"; exit 1; }
 
-# Generate archive files for each binary
-# $(1): the binary name to be archived
-define genProtoc
-.PHONY: gen-proto-$(1)
-gen-proto-$(1):
-	$(PROTOC) --go_out=. --go_opt=module=$(PROTO_PREFIX) --go-grpc_out=. --go-grpc_opt=require_unimplemented_servers=false,module=$(PROTO_PREFIX) ./proto/$(1)/*
-endef
-
-$(foreach ITEM,$(PROTOS),$(eval $(call genProtoc,$(ITEM))))
-
-GEN_PROTOS:=$(foreach ITEM,$(PROTOS),gen-proto-$(ITEM))
-
 .PHONY: gen-proto
-gen-proto: check-proto-version $(GEN_PROTOS) modtidy
+gen-proto: check-proto-version _gen-proto modtidy
+
+.PHONY: _gen-proto
+_gen-proto:
+	$(PROTOC) --go_out=. \
+	  --go_opt=module=$(PROTO_PREFIX) \
+	  --go-grpc_out=. \
+	  --go-grpc_opt=require_unimplemented_servers=false,module=$(PROTO_PREFIX) \
+	  ./proto/*.proto
 
 test:
 	go test -count 1 -timeout 300s --race ./...

README.md

@@ -13,36 +13,40 @@ Recurring jobs can have a TTL, a delayed start, expiry time, and a maximum numbe
 ## Getting started
 
 ```go
-import etcdcron "github.com/diagridio/go-etcd-cron"
-import "github.com/diagridio/go-etcd-cron/api"
-
-cron, err := etcdcron.New(Options{
-  Client:         client,
-  Namespace:      "abc",
-  PartitionID:    0,
-  PartitionTotal: 1,
-  TriggerFn: func(context.Context, *api.TriggerRequest) bool {
-    // Do something with your trigger here.
-    // Return true if the trigger was successful, false otherwise.
-    // Note, returning false will cause the job to be retried *immediately*.
-  	return true
-  },
-})
-if err != nil {
-  panic(err)
-}
-
-// TODO: Pass proper context and do something with returned error.
-go cron.Run(context.Background())
-
-payload, _ := anypb.New(wrapperspb.String("hello"))
-meta, _ := anypb.New(wrapperspb.String("world"))
-tt := time.Now().Add(time.Second).Format(time.RFC3339)
-
-cron.Add(ctx, "my-job", &api.Job{
-  DueTime:  &tt,
-  Payload:  payload,
-  Metadata: meta,
+import (
+  "github.com/diagridio/go-etcd-cron/api"
+  "github.com/diagridio/go-etcd-cron/cron"
+)
+
+func main() {
+  cron, err := cron.New(cron.Options{
+    Client:         client,
+    Namespace:      "abc",
+    PartitionID:    0,
+    PartitionTotal: 1,
+    TriggerFn: func(context.Context, *api.TriggerRequest) bool {
+      // Do something with your trigger here.
+      // Return true if the trigger was successful, false otherwise.
+      // Note, returning false will cause the job to be retried *immediately*.
+      return true
+    },
+  })
+  if err != nil {
+    panic(err)
+  }
+
+  // TODO: Pass proper context and do something with returned error.
+  go cron.Run(context.Background())
+
+  payload, _ := anypb.New(wrapperspb.String("hello"))
+  meta, _ := anypb.New(wrapperspb.String("world"))
+  tt := time.Now().Add(time.Second).Format(time.RFC3339)
+
+  cron.Add(context.TODO(), "my-job", &api.Job{
+    DueTime:  &tt,
+    Payload:  payload,
+    Metadata: meta,
+  })
 }
 ```
 

api.go

@@ -0,0 +1,52 @@
+/*
+Copyright (c) 2024 Diagrid Inc.
+Licensed under the MIT License.
+*/
+
+package api
+
+import (
+	"context"
+)
+
+// TriggerFunction is the type of the function that is called when a job is
+// triggered.
+// Returning true will "tick" the job forward to the next scheduled time.
+// Returning false will cause the job to be re-enqueued and triggered
+// immediately.
+type TriggerFunction func(context.Context, *TriggerRequest) bool
+
+// API is the interface for interacting with the cron instance.
+type API interface {
+	// Add adds a job to the cron instance.
+	Add(ctx context.Context, name string, job *Job) error
+
+	// Get gets a job from the cron instance.
+	Get(ctx context.Context, name string) (*Job, error)
+
+	// Delete deletes a job from the cron instance.
+	Delete(ctx context.Context, name string) error
+
+	// DeletePrefixes deletes all jobs with the given prefixes from the cron
+	// instance.
+	DeletePrefixes(ctx context.Context, prefixes ...string) error
+
+	// List lists all jobs under a given job name prefix.
+	List(ctx context.Context, prefix string) (*ListResponse, error)
+}
+
+// Interface is a cron interface. It schedules and manages job which are stored
+// and informed from ETCD. It uses a trigger function to call when a job is
+// triggered.
+// Jobs may be oneshot or recurring. Recurring jobs are scheduled to run at
+// their next scheduled time. Oneshot jobs are scheduled to run once and are
+// removed from the schedule after they are triggered.
+type Interface interface {
+	// Run is a blocking function that runs the cron instance. It will return an
+	// error if the instance is already running.
+	// Returns when the given context is cancelled, after doing all cleanup.
+	Run(ctx context.Context) error
+
+	// API implements the client API for the cron instance.
+	API
+}