Merge create callback and create subscrition (#681)

* Merge create callback and create subscrition

These will be separate api subsystem requests but the same
coroutine.

Author: dfarr

.gitignore

@@ -33,3 +33,4 @@ resonate.yml
 coverage.out
 dst*.html
 redoc-static.html
+api/index.html

Makefile

@@ -3,8 +3,6 @@ gen-proto:
 	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/promise_t.proto
 	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/promise.proto
 	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/callback_t.proto
-	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/callback.proto
-	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/subscription.proto
 	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/schedule_t.proto
 	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/schedule.proto
 	protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/app/subsystems/api/grpc/pb/lock.proto

api/openapi.yml

@@ -316,14 +316,19 @@ paths:
         404:
           description: Promise not found
 
-  # Callbacks
-  /callbacks:
+  /promises/callback/{id}:
     post:
       tags:
-        - Callbacks
-      summary: Create callback
-      operationId: createCallback
+        - Promises
+      summary: Create promise callback
+      operationId: createPromiseCallback
       parameters:
+        - in: path
+          name: id
+          required: true
+          description: The promise id
+          schema:
+            type: string
         - in: header
           name: request-id
           description: Unique tracking id
@@ -336,16 +341,10 @@ paths:
             schema:
               type: object
               required:
-                - id
-                - promiseId
                 - rootPromiseId
                 - timeout
                 - recv
               properties:
-                id:
-                  type: string
-                promiseId:
-                  type: string
                 rootPromiseId:
                   type: string
                 timeout:
@@ -377,14 +376,19 @@ paths:
         404:
           description: Promise not found
 
-  # Subscriptions
-  /subscriptions:
+  /promises/subscribe/{id}:
     post:
       tags:
-        - Subscription
-      summary: Create a Subscription
-      operationId: createSubscription
+        - Promises
+      summary: Create promise subscription
+      operationId: createPromiseSubscription
       parameters:
+        - in: path
+          name: id
+          required: true
+          description: The promise id
+          schema:
+            type: string
         - in: header
           name: request-id
           description: Unique tracking id
@@ -398,14 +402,11 @@ paths:
               type: object
               required:
                 - id
-                - promiseId
                 - timeout
                 - recv
               properties:
                 id:
                   type: string
-                promiseId:
-                  type: string
                 timeout:
                   type: integer
                   format: int64

cmd/callbacks/callbacks.go

@@ -15,6 +15,7 @@ import (
 	"github.com/resonatehq/resonate/internal/aio"
 	"github.com/resonatehq/resonate/internal/api"
 	"github.com/resonatehq/resonate/internal/app/coroutines"
+	"github.com/resonatehq/resonate/internal/app/validators"
 	"github.com/resonatehq/resonate/internal/kernel/system"
 	"github.com/resonatehq/resonate/internal/kernel/t_api"
 	"github.com/resonatehq/resonate/internal/metrics"
@@ -127,6 +128,9 @@ func RunDSTCmd() *cobra.Command {
 				aio.AddSubsystem(subsystem)
 			}
 
+			// api validators
+			api.AddValidator(t_api.CreateCallback, validators.CreateCallback)
+
 			// start api/aio
 			if err := api.Start(); err != nil {
 				return err
@@ -145,7 +149,6 @@ func RunDSTCmd() *cobra.Command {
 			system.AddOnRequest(t_api.CreatePromiseAndTask, coroutines.CreatePromiseAndTask)
 			system.AddOnRequest(t_api.CompletePromise, coroutines.CompletePromise)
 			system.AddOnRequest(t_api.CreateCallback, coroutines.CreateCallback)
-			system.AddOnRequest(t_api.CreateSubscription, coroutines.CreateSubscription)
 			system.AddOnRequest(t_api.ReadSchedule, coroutines.ReadSchedule)
 			system.AddOnRequest(t_api.SearchSchedules, coroutines.SearchSchedules)
 			system.AddOnRequest(t_api.CreateSchedule, coroutines.CreateSchedule)

cmd/dst/run.go

@@ -38,7 +38,7 @@ func setup(url, dest string) error {
 	if err := download(url, tmp); err != nil {
 		return err
 	}
-	defer util.DeferAndLog(func() error {return os.Remove(tmp)})
+	defer util.DeferAndLog(func() error { return os.Remove(tmp) })
 
 	if err := unzip(tmp, dest); err != nil {
 		return err

cmd/projects/scaffold.go

@@ -1,4 +1,4 @@
-package callbacks
+package promises
 
 import (
 	"context"
@@ -7,39 +7,39 @@ import (
 	"strings"
 	"time"
 
+	"github.com/resonatehq/resonate/internal/util"
 	"github.com/resonatehq/resonate/pkg/client"
 	v1 "github.com/resonatehq/resonate/pkg/client/v1"
 	"github.com/spf13/cobra"
 )
 
-var createCallbacksExample = `
+var createPromiseCallbackExample = `
 # Create a callback
-resonate callback create foo --promise-id bar --root-promise-id baz --timeout 1h --recv default
+resonate promises callback foo --root-promise-id bar --timeout 1h --recv default
 
 # Create a callback with url
-resonate callback create foo --promise-id bar --root-promise-id baz --timeout 1h --recv poll://default/6fa89b7e-4a56-40e8-ba4e-78864caa3278
+resonate promises callback foo --root-promise-id bar --timeout 1h --recv poll://default/1
 
 # Create a callback with object
-resonate callback create foo --promise-id bar --root-promise-id baz --timeout 1h --recv {"type": "poll", "data": {"group": "default", "id": "6fa89b7e-4a56-40e8-ba4e-78864caa3278"}}
+resonate promises callback foo --root-promise-id bar --timeout 1h --recv {"type": "poll", "data": {"group": "default", "id": "2"}}
 `
 
-func CreateCallbackCmd(c client.Client) *cobra.Command {
+func CreatePromiseCallbackCmd(c client.Client) *cobra.Command {
 	var (
-		promiseId     string
 		rootPromiseId string
-		timeout       time.Duration
 		recvStr       string
+		timeout       time.Duration
 	)
 	cmd := &cobra.Command{
-		Use:     "create <id>",
+		Use:     "callback <id>",
 		Short:   "Create callback",
-		Example: createCallbacksExample,
+		Example: createPromiseCallbackExample,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			if len(args) != 1 {
-				return errors.New("must specify an id")
+				return errors.New("must specify a promise id")
 			}
 
-			id := args[0]
+			promiseId := args[0]
 
 			var recv v1.Recv
 
@@ -58,26 +58,24 @@ func CreateCallbackCmd(c client.Client) *cobra.Command {
 				}
 			}
 
-			body := v1.CreateCallbackJSONRequestBody{
-				Id:            id,
-				PromiseId:     promiseId,
+			body := v1.CreatePromiseCallbackJSONRequestBody{
 				RootPromiseId: rootPromiseId,
-				Timeout:       time.Now().Add(timeout).UnixMilli(),
 				Recv:          recv,
+				Timeout:       time.Now().Add(timeout).UnixMilli(),
 			}
 
-			res, err := c.V1().CreateCallbackWithResponse(context.TODO(), nil, body)
+			res, err := c.V1().CreatePromiseCallbackWithResponse(context.TODO(), promiseId, nil, body)
 			if err != nil {
 				return err
 			}
 
 			if res.StatusCode() == 201 {
-				cmd.Printf("Created callback: %s\n", id)
+				cmd.Printf("Created callback: %s\n", util.ResumeId(rootPromiseId, promiseId))
 			} else if res.StatusCode() == 200 {
 				if res.JSON200.Promise != nil && res.JSON200.Promise.State != v1.PromiseStatePENDING {
 					cmd.Printf("Promise %s already %s\n", promiseId, strings.ToLower(string(res.JSON200.Promise.State)))
 				} else {
-					cmd.Printf("Created callback: %s (deduplicated)\n", id)
+					cmd.Printf("Created callback: %s (deduplicated)\n", util.ResumeId(rootPromiseId, promiseId))
 				}
 			} else {
 				cmd.PrintErrln(res.Status(), string(res.Body))
@@ -87,12 +85,10 @@ func CreateCallbackCmd(c client.Client) *cobra.Command {
 		},
 	}
 
-	cmd.Flags().StringVar(&promiseId, "promise-id", "", "promise id")
 	cmd.Flags().StringVar(&rootPromiseId, "root-promise-id", "", "root promise id")
-	cmd.Flags().DurationVar(&timeout, "timeout", 0, "task timeout")
 	cmd.Flags().StringVar(&recvStr, "recv", "default", "task receiver")
+	cmd.Flags().DurationVar(&timeout, "timeout", 0, "task timeout")
 
-	_ = cmd.MarkFlagRequired("promise-id")
 	_ = cmd.MarkFlagRequired("root-promise-id")
 	_ = cmd.MarkFlagRequired("timeout")
 

cmd/callbacks/create.go renamed to cmd/promises/callback.go

@@ -40,6 +40,8 @@ func NewCmd() *cobra.Command {
 	cmd.AddCommand(SearchPromisesCmd(c))
 	cmd.AddCommand(CreatePromiseCmd(c))
 	cmd.AddCommand(CompletePromiseCmds(c)...)
+	cmd.AddCommand(CreatePromiseCallbackCmd(c))
+	cmd.AddCommand(CreateSubscriptionCmd(c))
 
 	// Flags
 	cmd.PersistentFlags().StringVarP(&server, "server", "", "http://localhost:8001", "resonate url")

cmd/promises/promises.go

@@ -10,7 +10,7 @@ import (
 	"github.com/spf13/cobra"
 )
 
-var searchPromiseExample = `
+var searchPromisesExample = `
 # Search for all promises
 resonate promises search "*"
 
@@ -38,7 +38,7 @@ func SearchPromisesCmd(c client.Client) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:     "search <q>",
 		Short:   "Search promises",
-		Example: searchPromiseExample,
+		Example: searchPromisesExample,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			if len(args) != 1 {
 				return errors.New("must specify search query")