From 34588897d7541a674beec96803242066e887074c Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Tue, 12 Mar 2024 08:42:46 -0700
Subject: [PATCH 1/7] fix: support source and sink func declarations

---
 buildengine/build_go_test.go                  | 39 ++++++++++
 buildengine/build_kotlin_test.go              | 78 +++++++++++++++++++
 go-runtime/compile/schema.go                  |  3 -
 go-runtime/compile/schema_test.go             | 12 +++
 go-runtime/compile/testdata/one/one.go        | 19 +++++
 go-runtime/ftl/call.go                        | 30 ++++++-
 go.mod                                        |  4 +
 .../src/main/kotlin/xyz/block/ftl/Context.kt  |  8 ++
 .../test/kotlin/xyz/block/ftl/ContextTest.kt  | 10 ++-
 9 files changed, 196 insertions(+), 7 deletions(-)

diff --git a/buildengine/build_go_test.go b/buildengine/build_go_test.go
index 6ff522b5e..046ee376b 100644
--- a/buildengine/build_go_test.go
+++ b/buildengine/build_go_test.go
@@ -36,6 +36,23 @@ func TestGenerateGoModule(t *testing.T) {
 					Request:  &schema.Ref{Name: "EchoRequest"},
 					Response: &schema.Ref{Name: "EchoResponse"},
 				},
+				&schema.Data{Name: "SinkReq"},
+				&schema.Verb{
+					Name:     "sink",
+					Request:  &schema.DataRef{Name: "SinkReq"},
+					Response: &schema.Unit{},
+				},
+				&schema.Data{Name: "SourceResp"},
+				&schema.Verb{
+					Name:     "source",
+					Request:  &schema.Unit{},
+					Response: &schema.DataRef{Name: "SourceResp"},
+				},
+				&schema.Verb{
+					Name:     "nothing",
+					Request:  &schema.Unit{},
+					Response: &schema.Unit{},
+				},
 			}},
 			{Name: "test"},
 		},
@@ -47,6 +64,7 @@ package other
 
 import (
   "context"
+  "github.com/TBD54566975/ftl/go-runtime/ftl"
 )
 
 var _ = context.Background
@@ -77,6 +95,27 @@ type EchoResponse struct {
 func Echo(context.Context, EchoRequest) (EchoResponse, error) {
   panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
 }
+
+type SinkReq struct {
+}
+
+//ftl:verb
+func Sink(context.Context, SinkReq) (ftl.Unit, error) {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
+}
+
+type SourceResp struct {
+}
+
+//ftl:verb
+func Source(context.Context, ftl.Unit) (SourceResp, error) {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
+}
+
+//ftl:verb
+func Nothing(context.Context, ftl.Unit) (ftl.Unit, error) {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
+}
 `
 	bctx := buildContext{
 		moduleDir: "testdata/modules/another",
diff --git a/buildengine/build_kotlin_test.go b/buildengine/build_kotlin_test.go
index 904610a64..2cca7d216 100644
--- a/buildengine/build_kotlin_test.go
+++ b/buildengine/build_kotlin_test.go
@@ -305,3 +305,81 @@ fun emptyVerb(context: Context, req: ftl.builtin.Empty): ftl.builtin.Empty = thr
 		assertGeneratedModule("generated-sources/ftl/test/Test.kt", expected),
 	})
 }
+
+func TestGenerateSourcesAndSinks(t *testing.T) {
+	if testing.Short() {
+		t.SkipNow()
+	}
+	sch := &schema.Schema{
+		Modules: []*schema.Module{
+			schema.Builtins(),
+			{
+				Name: "test",
+				Decls: []schema.Decl{
+					&schema.Data{
+						Name: "SinkReq",
+						Fields: []*schema.Field{
+							{Name: "data", Type: &schema.Int{}},
+						}},
+					&schema.Verb{
+						Name:     "sink",
+						Request:  &schema.DataRef{Name: "SinkReq"},
+						Response: &schema.Unit{},
+					},
+					&schema.Data{
+						Name: "SourceResp",
+						Fields: []*schema.Field{
+							{Name: "data", Type: &schema.Int{}},
+						}},
+					&schema.Verb{
+						Name:     "source",
+						Request:  &schema.Unit{},
+						Response: &schema.DataRef{Name: "SourceResp"},
+					},
+					&schema.Verb{
+						Name:     "nothing",
+						Request:  &schema.Unit{},
+						Response: &schema.Unit{},
+					},
+				},
+			},
+		},
+	}
+
+	expected := `// Code generated by FTL. DO NOT EDIT.
+package ftl.test
+
+import xyz.block.ftl.Context
+import xyz.block.ftl.Ignore
+import xyz.block.ftl.Verb
+
+data class SinkReq(
+  val data: Long,
+)
+
+@Verb
+@Ignore
+fun sink(context: Context, req: SinkReq): Unit = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.call(::sink, ...)")
+data class SourceResp(
+  val data: Long,
+)
+
+@Verb
+@Ignore
+fun source(context: Context, req: Unit): SourceResp = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.call(::source, ...)")
+@Verb
+@Ignore
+fun nothing(context: Context, req: Unit): Unit = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.call(::nothing, ...)")
+`
+	bctx := buildContext{
+		moduleDir: "testdata/modules/echokotlin",
+		buildDir:  "target",
+		sch:       sch,
+	}
+	testBuild(t, bctx, []assertion{
+		assertGeneratedModule("generated-sources/ftl/test/Test.kt", expected),
+	})
+}
diff --git a/go-runtime/compile/schema.go b/go-runtime/compile/schema.go
index a11d833c2..36e729844 100644
--- a/go-runtime/compile/schema.go
+++ b/go-runtime/compile/schema.go
@@ -287,9 +287,6 @@ func checkSignature(sig *types.Signature) (req, resp *types.Var, err error) {
 		}
 		resp = results.At(0)
 	}
-	if params.Len() == 1 && results.Len() == 1 {
-		return nil, nil, fmt.Errorf("must either accept an input or return a result, but does neither")
-	}
 	return req, resp, nil
 }
 
diff --git a/go-runtime/compile/schema_test.go b/go-runtime/compile/schema_test.go
index df498c575..c22cc8df0 100644
--- a/go-runtime/compile/schema_test.go
+++ b/go-runtime/compile/schema_test.go
@@ -47,6 +47,12 @@ func TestExtractModuleSchema(t *testing.T) {
   data Resp {
   }
 
+  data SinkReq {
+  }
+
+  data SourceResp {
+  }
+
   enum Color(String) {
     Red("Red")
     Blue("Blue")
@@ -75,6 +81,12 @@ func TestExtractModuleSchema(t *testing.T) {
 
   secret secretValue String
 
+  verb nothing(Unit) Unit
+
+  verb sink(one.SinkReq) Unit
+
+  verb source(Unit) one.SourceResp
+
   verb verb(one.Req) one.Resp
 }
 `
diff --git a/go-runtime/compile/testdata/one/one.go b/go-runtime/compile/testdata/one/one.go
index 4d99fb3f0..4c8e63b68 100644
--- a/go-runtime/compile/testdata/one/one.go
+++ b/go-runtime/compile/testdata/one/one.go
@@ -79,3 +79,22 @@ func Verb(ctx context.Context, req Req) (Resp, error) {
 const Yellow Color = "Yellow"
 
 const YellowInt ColorInt = 3
+
+type SinkReq struct{}
+
+//ftl:verb
+func Sink(ctx context.Context, req SinkReq) error {
+	return nil
+}
+
+type SourceResp struct{}
+
+//ftl:verb
+func Source(ctx context.Context) (SourceResp, error) {
+	return SourceResp{}, nil
+}
+
+//ftl:verb
+func Nothing(ctx context.Context) error {
+	return nil
+}
diff --git a/go-runtime/ftl/call.go b/go-runtime/ftl/call.go
index 67988fa96..89a694d22 100644
--- a/go-runtime/ftl/call.go
+++ b/go-runtime/ftl/call.go
@@ -44,9 +44,26 @@ func Call[Req, Resp any](ctx context.Context, verb Verb[Req, Resp], req Req) (re
 	}
 }
 
-// Call a Sink through the FTL controller.
-// func CallSink[Req any](ctx context.Context, sink Sink[Req]) error {
-// }
+// CallSink calls a Sink through the FTL controller.
+func CallSink[Req any](ctx context.Context, sink Sink[Req], req Req) error {
+	verb := func(ctx context.Context, req Req) (Unit, error) {
+		err := sink(ctx, req)
+		return Unit{}, err
+	}
+
+	_, err := Call(ctx, verb, req)
+	return err
+}
+
+// CallSource calls a Source through the FTL controller.
+func CallSource[Resp any](ctx context.Context, source Source[Resp]) (Resp, error) {
+	verb := func(ctx context.Context, _ Unit) (Resp, error) {
+		resp, err := source(ctx)
+		return resp, err
+	}
+
+	return Call(ctx, verb, Unit{})
+}
 
 // VerbToRef returns the FTL reference for a Verb.
 func VerbToRef[Req, Resp any](verb Verb[Req, Resp]) Ref {
@@ -54,11 +71,18 @@ func VerbToRef[Req, Resp any](verb Verb[Req, Resp]) Ref {
 	return goRefToFTLRef(ref)
 }
 
+// SinkToRef returns the FTL reference for a Sink.
 func SinkToRef[Req any](sink Sink[Req]) Ref {
 	ref := runtime.FuncForPC(reflect.ValueOf(sink).Pointer()).Name()
 	return goRefToFTLRef(ref)
 }
 
+// SourceToRef returns the FTL reference for a Source.
+func SourceToRef[Resp any](source Source[Resp]) Ref {
+	ref := runtime.FuncForPC(reflect.ValueOf(source).Pointer()).Name()
+	return goRefToFTLRef(ref)
+}
+
 func goRefToFTLRef(ref string) Ref {
 	parts := strings.Split(ref[strings.LastIndex(ref, "/")+1:], ".")
 	return Ref{parts[len(parts)-2], strcase.ToLowerCamel(parts[len(parts)-1])}
diff --git a/go.mod b/go.mod
index 4cf08aa9a..1f8f4bdb7 100644
--- a/go.mod
+++ b/go.mod
@@ -33,6 +33,7 @@ require (
 	github.com/radovskyb/watcher v1.0.7
 	github.com/rs/cors v1.10.1
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
+	github.com/stretchr/testify v1.8.4
 	github.com/swaggest/jsonschema-go v0.3.69
 	github.com/titanous/json5 v1.0.0
 	github.com/tmc/langchaingo v0.1.5
@@ -57,10 +58,13 @@ require (
 
 require (
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
 	github.com/pkoukk/tiktoken-go v0.1.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.1.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 	modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect
 )
 
diff --git a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt
index 76d7c6d21..829fb9c33 100644
--- a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt
+++ b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt
@@ -28,4 +28,12 @@ class Context(
     val responseJson = routingClient.call(this, Ref(ftlModule, verb.name), requestJson)
     return gson.fromJson(responseJson, R::class.java)
   }
+
+  inline fun <reified R> callSink(verb: KFunction<R>, request: Any) {
+    call(verb, request)
+  }
+
+  inline fun <reified R> callSource(verb: KFunction<R>): R {
+    return call(verb, Unit)
+  }
 }
diff --git a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt
index c1ae8c492..d318e85b6 100644
--- a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt
+++ b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt
@@ -23,6 +23,14 @@ class ContextTest {
           invoke = { ctx -> ctx.call(::time, Empty()) },
           expected = TimeResponse(staticTime),
         ),
+        TestCase(
+          invoke = { ctx -> ctx.callSink(::time, EchoRequest("Alice")) },
+          expected = Unit,
+        ),
+        TestCase(
+          invoke = { ctx -> ctx.callSource(::time) },
+          expected = TimeResponse(staticTime),
+        ),
       )
     }
   }
@@ -35,6 +43,6 @@ class ContextTest {
     val routingClient = LoopbackVerbServiceClient(registry)
     val context = Context("ftl.test", routingClient)
     val result = testCase.invoke(context)
-    assertEquals(result, testCase.expected)
+    assertEquals(testCase.expected, result)
   }
 }

From 1a523d5d1526c6eaf6ccfd4ca7a710a1b31d5bab Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Wed, 13 Mar 2024 11:39:30 -0700
Subject: [PATCH 2/7] working calls to source, sink, empty in go

---
 buildengine/build_go_test.go                  | 13 +++---
 examples/go/echo/echo.go                      | 12 +++++
 examples/go/time/time.go                      | 15 +++++++
 .../build-template/_ftl.tmpl/go/main/main.go  | 18 +++++---
 go-runtime/compile/build.go                   | 19 ++++++--
 .../external_module.go                        | 14 ++++++
 go-runtime/compile/schema.go                  | 12 ++++-
 go-runtime/ftl/call.go                        | 45 ++++++++++++-------
 go-runtime/server/server.go                   | 45 +++++++++++++++++--
 9 files changed, 158 insertions(+), 35 deletions(-)

diff --git a/buildengine/build_go_test.go b/buildengine/build_go_test.go
index 046ee376b..6b65f6b4a 100644
--- a/buildengine/build_go_test.go
+++ b/buildengine/build_go_test.go
@@ -64,7 +64,6 @@ package other
 
 import (
   "context"
-  "github.com/TBD54566975/ftl/go-runtime/ftl"
 )
 
 var _ = context.Background
@@ -100,21 +99,21 @@ type SinkReq struct {
 }
 
 //ftl:verb
-func Sink(context.Context, SinkReq) (ftl.Unit, error) {
-  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
+func Sink(context.Context, SinkReq) error {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.CallSink()")
 }
 
 type SourceResp struct {
 }
 
 //ftl:verb
-func Source(context.Context, ftl.Unit) (SourceResp, error) {
-  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
+func Source(context.Context) (SourceResp, error) {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.CallSource()")
 }
 
 //ftl:verb
-func Nothing(context.Context, ftl.Unit) (ftl.Unit, error) {
-  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
+func Nothing(context.Context) error {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.CallEmpty()")
 }
 `
 	bctx := buildContext{
diff --git a/examples/go/echo/echo.go b/examples/go/echo/echo.go
index a831639c3..db74e77f9 100644
--- a/examples/go/echo/echo.go
+++ b/examples/go/echo/echo.go
@@ -32,5 +32,17 @@ func Echo(ctx context.Context, req EchoRequest) (EchoResponse, error) {
 	if err != nil {
 		return EchoResponse{}, err
 	}
+	err = ftl.CallEmpty(ctx, time.Empty)
+	if err != nil {
+		return EchoResponse{}, err
+	}
+	err = ftl.CallSink(ctx, time.Sink, time.TimeRequest{})
+	if err != nil {
+		return EchoResponse{}, err
+	}
+	tresp, err = ftl.CallSource(ctx, time.Source)
+	if err != nil {
+		return EchoResponse{}, err
+	}
 	return EchoResponse{Message: fmt.Sprintf("Hello, %s!!! It is %s!", req.Name.Default(defaultName.Get(ctx)), tresp.Time)}, nil
 }
diff --git a/examples/go/time/time.go b/examples/go/time/time.go
index ed9ff61ed..4eee0686e 100644
--- a/examples/go/time/time.go
+++ b/examples/go/time/time.go
@@ -17,3 +17,18 @@ type TimeResponse struct {
 func Time(ctx context.Context, req TimeRequest) (TimeResponse, error) {
 	return TimeResponse{Time: time.Now()}, nil
 }
+
+//ftl:verb
+func Sink(ctx context.Context, req TimeRequest) error {
+	return nil
+}
+
+//ftl:verb
+func Source(ctx context.Context) (TimeResponse, error) {
+	return TimeResponse{Time: time.Now()}, nil
+}
+
+//ftl:verb
+func Empty(ctx context.Context) error {
+	return nil
+}
diff --git a/go-runtime/compile/build-template/_ftl.tmpl/go/main/main.go b/go-runtime/compile/build-template/_ftl.tmpl/go/main/main.go
index c96199838..60103dddb 100644
--- a/go-runtime/compile/build-template/_ftl.tmpl/go/main/main.go
+++ b/go-runtime/compile/build-template/_ftl.tmpl/go/main/main.go
@@ -12,10 +12,18 @@ import (
 )
 
 func main() {
-  verbConstructor := server.NewUserVerbServer("{{.Name}}",
+	verbConstructor := server.NewUserVerbServer("{{.Name}}",
 {{- range .Verbs}}
-    server.Handle({{$.Name}}.{{.Name}}),
+	{{- if and .HasRequest .HasResponse}}
+		server.HandleCall({{$.Name}}.{{.Name}}),
+	{{- else if .HasRequest}}
+		server.HandleSink({{$.Name}}.{{.Name}}),
+	{{- else if .HasResponse}}
+		server.HandleSource({{$.Name}}.{{.Name}}),
+	{{- else}}
+		server.HandleEmpty({{$.Name}}.{{.Name}}),
+	{{- end}}
 {{- end}}
-  )
-  plugin.Start(context.Background(), "{{.Name}}", verbConstructor, ftlv1connect.VerbServiceName, ftlv1connect.NewVerbServiceHandler)
-}
\ No newline at end of file
+	)
+	plugin.Start(context.Background(), "{{.Name}}", verbConstructor, ftlv1connect.VerbServiceName, ftlv1connect.NewVerbServiceHandler)
+}
diff --git a/go-runtime/compile/build.go b/go-runtime/compile/build.go
index c00a5e411..c64f81103 100644
--- a/go-runtime/compile/build.go
+++ b/go-runtime/compile/build.go
@@ -34,7 +34,9 @@ type externalModuleContext struct {
 }
 
 type goVerb struct {
-	Name string
+	Name        string
+	HasRequest  bool
+	HasResponse bool
 }
 
 type mainModuleContext struct {
@@ -122,7 +124,15 @@ func Build(ctx context.Context, moduleDir string, sch *schema.Schema) error {
 			if !ok {
 				return fmt.Errorf("missing native name for verb %s", verb.Name)
 			}
-			goVerbs = append(goVerbs, goVerb{Name: nativeName})
+
+			goverb := goVerb{Name: nativeName}
+			if _, ok := verb.Request.(*schema.Unit); !ok {
+				goverb.HasRequest = true
+			}
+			if _, ok := verb.Response.(*schema.Unit); !ok {
+				goverb.HasResponse = true
+			}
+			goVerbs = append(goVerbs, goverb)
 		}
 	}
 	if err := internal.ScaffoldZip(buildTemplateFiles(), moduleDir, mainModuleContext{
@@ -183,7 +193,7 @@ var scaffoldFuncs = scaffolder.FuncMap{
 			case *schema.Time:
 				imports["time"] = "stdtime"
 
-			case *schema.Optional, *schema.Unit:
+			case *schema.Optional:
 				imports["github.com/TBD54566975/ftl/go-runtime/ftl"] = ""
 
 			default:
@@ -201,6 +211,9 @@ var scaffoldFuncs = scaffolder.FuncMap{
 		}
 		panic(fmt.Sprintf("unsupported value %T", v))
 	},
+	"isSink": func(v schema.Verb) string {
+		return "wes"
+	},
 }
 
 func genType(module *schema.Module, t schema.Type) string {
diff --git a/go-runtime/compile/external-module-template/_ftl/go/modules/{{ range .NonMainModules }}{{ push .Name . }}{{ end }}/external_module.go b/go-runtime/compile/external-module-template/_ftl/go/modules/{{ range .NonMainModules }}{{ push .Name . }}{{ end }}/external_module.go
index 55ba14072..22713846a 100644
--- a/go-runtime/compile/external-module-template/_ftl/go/modules/{{ range .NonMainModules }}{{ push .Name . }}{{ end }}/external_module.go	
+++ b/go-runtime/compile/external-module-template/_ftl/go/modules/{{ range .NonMainModules }}{{ push .Name . }}{{ end }}/external_module.go	
@@ -38,8 +38,22 @@ type {{.Name|title}}
 {{if .Comments}}//
 {{end -}}
 //ftl:verb
+{{- if and (eq (type $ .Request) "ftl.Unit") (eq (type $ .Response) "ftl.Unit")}}
+func {{.Name|title}}(context.Context) error {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.CallEmpty()")
+}
+{{- else if eq (type $ .Request) "ftl.Unit"}}
+func {{.Name|title}}(context.Context) ({{type $ .Response}}, error) {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.CallSource()")
+}
+{{- else if eq (type $ .Response) "ftl.Unit"}}
+func {{.Name|title}}(context.Context, {{type $ .Request}}) error {
+  panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.CallSink()")
+}
+{{- else}}
 func {{.Name|title}}(context.Context, {{type $ .Request}}) ({{type $ .Response}}, error) {
   panic("Verb stubs should not be called directly, instead use github.com/TBD54566975/ftl/runtime-go/ftl.Call()")
 }
 {{- end}}
 {{- end}}
+{{- end}}
diff --git a/go-runtime/compile/schema.go b/go-runtime/compile/schema.go
index 36e729844..418878b1b 100644
--- a/go-runtime/compile/schema.go
+++ b/go-runtime/compile/schema.go
@@ -30,11 +30,12 @@ var (
 	errorIFaceType = once(func() *types.Interface {
 		return mustLoadRef("builtin", "error").Type().Underlying().(*types.Interface) //nolint:forcetypeassert
 	})
+
 	ftlCallFuncPath   = "github.com/TBD54566975/ftl/go-runtime/ftl.Call"
 	ftlConfigFuncPath = "github.com/TBD54566975/ftl/go-runtime/ftl.Config"
 	ftlSecretFuncPath = "github.com/TBD54566975/ftl/go-runtime/ftl.Secret" //nolint:gosec
-
-	aliasFieldTag = "json"
+	ftlUnitTypePath   = "github.com/TBD54566975/ftl/go-runtime/ftl.Unit"
+	aliasFieldTag     = "json"
 )
 
 // NativeNames is a map of top-level declarations to their native Go names.
@@ -269,6 +270,10 @@ func checkSignature(sig *types.Signature) (req, resp *types.Var, err error) {
 		if !isType[*types.Struct](params.At(1).Type()) {
 			return nil, nil, fmt.Errorf("second parameter must be a struct but is %s", params.At(1).Type())
 		}
+		if params.At(1).Type().String() == ftlUnitTypePath {
+			return nil, nil, fmt.Errorf("second parameter must not be ftl.Unit")
+		}
+
 		req = params.At(1)
 	}
 
@@ -285,6 +290,9 @@ func checkSignature(sig *types.Signature) (req, resp *types.Var, err error) {
 		if !isType[*types.Struct](results.At(0).Type()) {
 			return nil, nil, fmt.Errorf("first result must be a struct but is %s", results.At(0).Type())
 		}
+		if results.At(1).Type().String() == ftlUnitTypePath {
+			return nil, nil, fmt.Errorf("first result must not be ftl.Unit")
+		}
 		resp = results.At(0)
 	}
 	return req, resp, nil
diff --git a/go-runtime/ftl/call.go b/go-runtime/ftl/call.go
index 89a694d22..9f7f04c74 100644
--- a/go-runtime/ftl/call.go
+++ b/go-runtime/ftl/call.go
@@ -11,20 +11,19 @@ import (
 
 	ftlv1 "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1"
 	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
+	schemapb "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/schema"
 	"github.com/TBD54566975/ftl/backend/schema/strcase"
 	"github.com/TBD54566975/ftl/go-runtime/encoding"
 	"github.com/TBD54566975/ftl/internal/rpc"
 )
 
-// Call a Verb through the FTL Controller.
-func Call[Req, Resp any](ctx context.Context, verb Verb[Req, Resp], req Req) (resp Resp, err error) {
-	callee := VerbToRef(verb)
+func call[Req, Resp any](ctx context.Context, callee *schemapb.VerbRef, req Req) (resp Resp, err error) {
 	client := rpc.ClientFromContext[ftlv1connect.VerbServiceClient](ctx)
 	reqData, err := encoding.Marshal(req)
 	if err != nil {
 		return resp, fmt.Errorf("%s: failed to marshal request: %w", callee, err)
 	}
-	cresp, err := client.Call(ctx, connect.NewRequest(&ftlv1.CallRequest{Verb: callee.ToProto(), Body: reqData}))
+	cresp, err := client.Call(ctx, connect.NewRequest(&ftlv1.CallRequest{Verb: callee, Body: reqData}))
 	if err != nil {
 		return resp, fmt.Errorf("%s: failed to call Verb: %w", callee, err)
 	}
@@ -44,25 +43,35 @@ func Call[Req, Resp any](ctx context.Context, verb Verb[Req, Resp], req Req) (re
 	}
 }
 
+// Call a Verb through the FTL Controller.
+func Call[Req, Resp any](ctx context.Context, verb Verb[Req, Resp], req Req) (resp Resp, err error) {
+	callee := VerbToRef(verb)
+	return call[Req, Resp](ctx, callee.ToProto(), req)
+}
+
 // CallSink calls a Sink through the FTL controller.
 func CallSink[Req any](ctx context.Context, sink Sink[Req], req Req) error {
-	verb := func(ctx context.Context, req Req) (Unit, error) {
-		err := sink(ctx, req)
-		return Unit{}, err
-	}
-
-	_, err := Call(ctx, verb, req)
+	callee := SinkToRef(sink)
+	verbRef := &schemapb.VerbRef{Module: callee.Module, Name: callee.Name}
+	_, err := call[Req, Unit](ctx, verbRef, req)
 	return err
 }
 
 // CallSource calls a Source through the FTL controller.
 func CallSource[Resp any](ctx context.Context, source Source[Resp]) (Resp, error) {
-	verb := func(ctx context.Context, _ Unit) (Resp, error) {
-		resp, err := source(ctx)
-		return resp, err
-	}
+	callee := SourceToRef(source)
+	verbRef := &schemapb.VerbRef{Module: callee.Module, Name: callee.Name}
+	fmt.Printf("source ref2: %v\n", verbRef)
+	return call[Unit, Resp](ctx, verbRef, Unit{})
+}
+
+// CallEmpty calls a Verb with no request or response through the FTL controller.
+func CallEmpty(ctx context.Context, empty Empty) error {
+	callee := EmptyToRef(empty)
+	verbRef := &schemapb.VerbRef{Module: callee.Module, Name: callee.Name}
 
-	return Call(ctx, verb, Unit{})
+	_, err := call[Unit, Unit](ctx, verbRef, Unit{})
+	return err
 }
 
 // VerbToRef returns the FTL reference for a Verb.
@@ -83,6 +92,12 @@ func SourceToRef[Resp any](source Source[Resp]) Ref {
 	return goRefToFTLRef(ref)
 }
 
+// EmptyToRef returns the FTL reference for an Empty.
+func EmptyToRef(empty Empty) Ref {
+	ref := runtime.FuncForPC(reflect.ValueOf(empty).Pointer()).Name()
+	return goRefToFTLRef(ref)
+}
+
 func goRefToFTLRef(ref string) Ref {
 	parts := strings.Split(ref[strings.LastIndex(ref, "/")+1:], ".")
 	return Ref{parts[len(parts)-2], strcase.ToLowerCamel(parts[len(parts)-1])}
diff --git a/go-runtime/server/server.go b/go-runtime/server/server.go
index 747bab95e..ecbd7a3b7 100644
--- a/go-runtime/server/server.go
+++ b/go-runtime/server/server.go
@@ -65,9 +65,7 @@ type Handler struct {
 	fn  func(ctx context.Context, req []byte) ([]byte, error)
 }
 
-// Handle creates a Handler from a Verb.
-func Handle[Req, Resp any](verb func(ctx context.Context, req Req) (Resp, error)) Handler {
-	ref := ftl.VerbToRef(verb)
+func handler[Req, Resp any](ref ftl.VerbRef, verb func(ctx context.Context, req Req) (Resp, error)) Handler {
 	return Handler{
 		ref: ref,
 		fn: func(ctx context.Context, reqdata []byte) ([]byte, error) {
@@ -94,6 +92,47 @@ func Handle[Req, Resp any](verb func(ctx context.Context, req Req) (Resp, error)
 	}
 }
 
+// HandleCall creates a Handler from a Verb.
+func HandleCall[Req, Resp any](verb func(ctx context.Context, req Req) (Resp, error)) Handler {
+	return handler(ftl.VerbToRef(verb), verb)
+}
+
+// HandleSink creates a Handler from a Sink with no response.
+func HandleSink[Req any](sink func(ctx context.Context, req Req) error) Handler {
+	ref := ftl.SinkToRef(sink)
+
+	verb := func(ctx context.Context, req Req) (ftl.Unit, error) {
+		err := sink(ctx, req)
+		return ftl.Unit{}, err
+	}
+
+	return handler(ftl.VerbRef{Module: ref.Module, Name: ref.Name}, verb)
+}
+
+// HandleSource creates a Handler from a Source with no request.
+func HandleSource[Resp any](source func(ctx context.Context) (Resp, error)) Handler {
+	ref := ftl.SourceToRef(source)
+
+	fmt.Printf("source ref: %v\n", ref)
+	verb := func(ctx context.Context, _ ftl.Unit) (Resp, error) {
+		return source(ctx)
+	}
+
+	return handler(ftl.VerbRef{Module: ref.Module, Name: ref.Name}, verb)
+}
+
+// HandleEmpty creates a Handler from a Verb with no request or response.
+func HandleEmpty(empty func(ctx context.Context) error) Handler {
+	ref := ftl.EmptyToRef(empty)
+
+	verb := func(ctx context.Context, _ ftl.Unit) (ftl.Unit, error) {
+		err := empty(ctx)
+		return ftl.Unit{}, err
+	}
+
+	return handler(ftl.VerbRef{Module: ref.Module, Name: ref.Name}, verb)
+}
+
 var _ ftlv1connect.VerbServiceHandler = (*moduleServer)(nil)
 
 // This is the server that is compiled into the same binary as user-defined Verbs.

From e60fd42dba051f85ef03d457d2cca95551c6fd8d Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Wed, 13 Mar 2024 12:59:06 -0700
Subject: [PATCH 3/7] Add kotlin updates

---
 buildengine/build_kotlin_test.go              | 10 +--
 .../{{ .Name | camel }}.kt                    | 11 +++
 .../src/main/kotlin/xyz/block/ftl/Context.kt  |  4 +
 .../ftl/schemaextractor/ExtractSchemaRule.kt  | 55 +++++++-----
 .../test/kotlin/xyz/block/ftl/ContextTest.kt  |  4 +
 .../schemaextractor/ExtractSchemaRuleTest.kt  | 83 +++++++++++++++++++
 6 files changed, 142 insertions(+), 25 deletions(-)

diff --git a/buildengine/build_kotlin_test.go b/buildengine/build_kotlin_test.go
index 2cca7d216..94b0db764 100644
--- a/buildengine/build_kotlin_test.go
+++ b/buildengine/build_kotlin_test.go
@@ -360,19 +360,19 @@ data class SinkReq(
 @Verb
 @Ignore
 fun sink(context: Context, req: SinkReq): Unit = throw
-    NotImplementedError("Verb stubs should not be called directly, instead use context.call(::sink, ...)")
+    NotImplementedError("Verb stubs should not be called directly, instead use context.callSink(::sink, ...)")
 data class SourceResp(
   val data: Long,
 )
 
 @Verb
 @Ignore
-fun source(context: Context, req: Unit): SourceResp = throw
-    NotImplementedError("Verb stubs should not be called directly, instead use context.call(::source, ...)")
+fun source(context: Context): SourceResp = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.callSource(::source, ...)")
 @Verb
 @Ignore
-fun nothing(context: Context, req: Unit): Unit = throw
-    NotImplementedError("Verb stubs should not be called directly, instead use context.call(::nothing, ...)")
+fun nothing(context: Context): Unit = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.callEmpty(::nothing, ...)")
 `
 	bctx := buildContext{
 		moduleDir: "testdata/modules/echokotlin",
diff --git a/kotlin-runtime/external-module-template/target.tmpl/generated-sources/ftl/{{ range .ExternalModules }}{{ push .Name . }}{{ end }}/{{ .Name | camel }}.kt b/kotlin-runtime/external-module-template/target.tmpl/generated-sources/ftl/{{ range .ExternalModules }}{{ push .Name . }}{{ end }}/{{ .Name | camel }}.kt
index 86453b047..577c2fa2e 100644
--- a/kotlin-runtime/external-module-template/target.tmpl/generated-sources/ftl/{{ range .ExternalModules }}{{ push .Name . }}{{ end }}/{{ .Name | camel }}.kt	
+++ b/kotlin-runtime/external-module-template/target.tmpl/generated-sources/ftl/{{ range .ExternalModules }}{{ push .Name . }}{{ end }}/{{ .Name | camel }}.kt	
@@ -34,8 +34,19 @@ data class {{.Name|title}}
 {{- else if is "Verb" . }}
 {{.Comments|comment -}}@Verb
 @Ignore
+{{- if and (eq (type $ .Request) "Unit") (eq (type $ .Response) "Unit")}}
+fun {{.Name|lowerCamel}}(context: Context): Unit = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.callEmpty(::{{.Name|lowerCamel}}, ...)")
+{{- else if eq (type $ .Request) "Unit"}}
+fun {{.Name|lowerCamel}}(context: Context): {{type $ .Response}} = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.callSource(::{{.Name|lowerCamel}}, ...)")
+{{- else if eq (type $ .Response) "Unit"}}
+fun {{.Name|lowerCamel}}(context: Context, req: {{type $ .Request}}): Unit = throw
+    NotImplementedError("Verb stubs should not be called directly, instead use context.callSink(::{{.Name|lowerCamel}}, ...)")
+{{- else}}
 fun {{.Name|lowerCamel}}(context: Context, req: {{type $ .Request}}): {{type $ .Response}} = throw
     NotImplementedError("Verb stubs should not be called directly, instead use context.call(::{{.Name|lowerCamel}}, ...)")
 {{- end}}
+{{- end}}
 
 {{- end}}
diff --git a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt
index 829fb9c33..0041ed4ba 100644
--- a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt
+++ b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/Context.kt
@@ -36,4 +36,8 @@ class Context(
   inline fun <reified R> callSource(verb: KFunction<R>): R {
     return call(verb, Unit)
   }
+
+  inline fun <reified R> callEmpty(verb: KFunction<R>) {
+    call(verb, Unit)
+  }
 }
diff --git a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
index 50f271621..9f91df1ec 100644
--- a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
+++ b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
@@ -209,44 +209,55 @@ class SchemaExtractor(
       }
 
       // Validate parameters
-      require(verb.valueParameters.size == 2) { "$verbSourcePos Verbs must have exactly two arguments, ${verb.name} did not" }
+      require(verb.valueParameters.size >= 1) { "$verbSourcePos Verbs must have at least one argument, ${verb.name} did not" }
+      require(verb.valueParameters.size <= 2) { "$verbSourcePos Verbs must have at most two arguments, ${verb.name} did not" }
       val ctxParam = verb.valueParameters.first()
-      val reqParam = verb.valueParameters.last()
+
       require(ctxParam.typeReference?.resolveType()?.fqNameOrNull()?.asString() == Context::class.qualifiedName) {
         "${verb.valueParameters.first().getLineAndColumn()} First argument of verb must be Context"
       }
 
-      require(reqParam.typeReference?.resolveType()
-        ?.let { it.toClassDescriptor().isData || it.isEmptyBuiltin() }
-        ?: false
-      ) {
-        "${verb.valueParameters.last().getLineAndColumn()} Second argument of ${verb.name} must be a data class or " +
-          "builtin.Empty"
+      if (verb.valueParameters.size == 2) {
+        val reqParam = verb.valueParameters.last()
+        require(reqParam.typeReference?.resolveType()
+          ?.let { it.toClassDescriptor().isData || it.isEmptyBuiltin() || it.isUnit() }
+          ?: false
+        ) {
+          "${verb.valueParameters.last().getLineAndColumn()} Second argument of ${verb.name} must be a data class or " +
+            "builtin.Empty"
+        }
       }
 
       // Validate return type
-      val respClass = verb.createTypeBindingForReturnType(bindingContext)?.type
-        ?: throw IllegalStateException("$verbSourcePos Could not resolve ${verb.name} return type")
-      require(respClass.toClassDescriptor().isData || respClass.isEmptyBuiltin()) {
-        "${verbSourcePos}: return type of ${verb.name} must be a data class or builtin.Empty but is ${
-          respClass.fqNameOrNull()?.asString()
-        }"
+      verb.createTypeBindingForReturnType(bindingContext)?.type?.let {
+        require(it.toClassDescriptor().isData || it.isEmptyBuiltin() || it.isUnit()) {
+          "${verbSourcePos}: return type of ${verb.name} must be a data class or builtin.Empty but is ${
+            it.fqNameOrNull()?.asString()
+          }"
+        }
       }
     }
   }
 
   private fun extractVerb(verb: KtNamedFunction): Verb {
     val verbSourcePos = verb.getLineAndColumn()
-    val requestRef = verb.valueParameters.last()?.let {
-      val position = it.getLineAndColumn().toPosition()
-      return@let it.typeReference?.resolveType()?.toSchemaType(position)
+
+    var requestRef = Type(unit = xyz.block.ftl.v1.schema.Unit())
+    if (verb.valueParameters.size > 1) {
+      val ref = verb.valueParameters.last()?.let {
+        val position = it.getLineAndColumn().toPosition()
+        return@let it.typeReference?.resolveType()?.toSchemaType(position)
+      }
+      ref?.let { requestRef = it }
+
     }
-    requireNotNull(requestRef) { "$verbSourcePos Could not resolve request type for ${verb.name}" }
-    val returnRef = verb.createTypeBindingForReturnType(bindingContext)?.let {
+
+    var returnRef = Type(unit = xyz.block.ftl.v1.schema.Unit())
+    val ref = verb.createTypeBindingForReturnType(bindingContext)?.let {
       val position = it.psiElement.getLineAndColumn().toPosition()
       return@let it.type.toSchemaType(position)
     }
-    requireNotNull(returnRef) { "$verbSourcePos Could not resolve response type for ${verb.name}" }
+    ref?.let { returnRef = it }
 
     val metadata = mutableListOf<Metadata>()
     extractIngress(verb, requestRef, returnRef)?.apply { metadata.add(Metadata(ingress = this)) }
@@ -672,6 +683,10 @@ class SchemaExtractor(
     private fun KotlinType.isEmptyBuiltin(): Boolean {
       return this.fqNameOrNull()?.asString() == "ftl.builtin.Empty"
     }
+
+    private fun KotlinType.isUnit(): Boolean {
+      return this.fqNameOrNull()?.asString() == "kotlin.Unit"
+    }
   }
 }
 
diff --git a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt
index d318e85b6..984be21f8 100644
--- a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt
+++ b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/ContextTest.kt
@@ -31,6 +31,10 @@ class ContextTest {
           invoke = { ctx -> ctx.callSource(::time) },
           expected = TimeResponse(staticTime),
         ),
+        TestCase(
+          invoke = { ctx -> ctx.callEmpty(::time) },
+          expected = Unit,
+        ),
       )
     }
   }
diff --git a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt
index 7e842f84f..167a04fe4 100644
--- a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt
+++ b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt
@@ -111,6 +111,15 @@ internal class ExtractSchemaRuleTest(private val env: KotlinCoreEnvironment) {
         //context.call(::foo, builtin.Empty())
         return context.call(::verb, builtin.Empty())
       }
+
+      @Verb
+      fun sink(context: Context, req: Empty) {}
+
+      @Verb
+      fun source(context: Context): Empty {}
+
+      @Verb
+      fun emptyVerb(context: Context) {}
     """
     ExtractSchemaRule(Config.empty).compileAndLintWithContext(env, code)
     val file = File(OUTPUT_FILENAME)
@@ -295,6 +304,45 @@ internal class ExtractSchemaRuleTest(private val env: KotlinCoreEnvironment) {
               )
             ),
           ),
+        ),
+        Decl(
+          verb = Verb(
+            name = "sink",
+            request = Type(
+              dataRef = DataRef(
+                name = "Empty",
+                module = "builtin"
+              )
+            ),
+            response = Type(
+              unit = Unit()
+            ),
+          ),
+        ),
+        Decl(
+          verb = Verb(
+            name = "source",
+            request = Type(
+              unit = Unit()
+            ),
+            response = Type(
+              dataRef = DataRef(
+                name = "Empty",
+                module = "builtin"
+              )
+            ),
+          ),
+        ),
+        Decl(
+          verb = Verb(
+            name = "emptyVerb",
+            request = Type(
+              unit = Unit()
+            ),
+            response = Type(
+              unit = Unit()
+            ),
+          ),
         )
       )
     )
@@ -370,6 +418,41 @@ import xyz.block.ftl.Verb
 data class EchoRequest(val name: String)
 data class EchoResponse(val message: String)
 
+/**
+ * Echoes the given message.
+ */
+@Throws(InvalidInput::class)
+@Verb
+@HttpIngress(Method.GET, "/echo")
+fun echo(context: Context, req: EchoRequest): EchoResponse {
+  return EchoResponse(messages = listOf(EchoMessage(message = "Hello!")))
+}
+        """
+    val message = assertThrows<java.lang.IllegalArgumentException> {
+      ExtractSchemaRule(Config.empty).compileAndLintWithContext(env, code)
+    }.message!!
+    assertContains(message, "@HttpIngress-annotated echo request must be ftl.builtin.HttpRequest")
+  }
+
+  @Test
+  fun `source and sink types`() {
+    val code = """
+ /**
+ * Echo module.
+ */
+package ftl.echo
+
+import xyz.block.ftl.Context
+import xyz.block.ftl.HttpIngress
+import xyz.block.ftl.Method
+import xyz.block.ftl.Verb
+
+/**
+ * Request to echo a message.
+ */
+data class EchoRequest(val name: String)
+data class EchoResponse(val message: String)
+
 /**
  * Echoes the given message.
  */

From b12b29b379c8436e73a61cdcc26f24f20c5487ef Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Wed, 13 Mar 2024 13:44:15 -0700
Subject: [PATCH 4/7] remove echo and time changes

---
 examples/go/echo/echo.go | 14 +-------------
 examples/go/time/time.go | 15 ---------------
 2 files changed, 1 insertion(+), 28 deletions(-)

diff --git a/examples/go/echo/echo.go b/examples/go/echo/echo.go
index db74e77f9..8c8fa0313 100644
--- a/examples/go/echo/echo.go
+++ b/examples/go/echo/echo.go
@@ -27,22 +27,10 @@ type EchoResponse struct {
 //
 //ftl:verb
 func Echo(ctx context.Context, req EchoRequest) (EchoResponse, error) {
-	fmt.Println("Echo received a request!")
 	tresp, err := ftl.Call(ctx, time.Time, time.TimeRequest{})
 	if err != nil {
 		return EchoResponse{}, err
 	}
-	err = ftl.CallEmpty(ctx, time.Empty)
-	if err != nil {
-		return EchoResponse{}, err
-	}
-	err = ftl.CallSink(ctx, time.Sink, time.TimeRequest{})
-	if err != nil {
-		return EchoResponse{}, err
-	}
-	tresp, err = ftl.CallSource(ctx, time.Source)
-	if err != nil {
-		return EchoResponse{}, err
-	}
+
 	return EchoResponse{Message: fmt.Sprintf("Hello, %s!!! It is %s!", req.Name.Default(defaultName.Get(ctx)), tresp.Time)}, nil
 }
diff --git a/examples/go/time/time.go b/examples/go/time/time.go
index 4eee0686e..ed9ff61ed 100644
--- a/examples/go/time/time.go
+++ b/examples/go/time/time.go
@@ -17,18 +17,3 @@ type TimeResponse struct {
 func Time(ctx context.Context, req TimeRequest) (TimeResponse, error) {
 	return TimeResponse{Time: time.Now()}, nil
 }
-
-//ftl:verb
-func Sink(ctx context.Context, req TimeRequest) error {
-	return nil
-}
-
-//ftl:verb
-func Source(ctx context.Context) (TimeResponse, error) {
-	return TimeResponse{Time: time.Now()}, nil
-}
-
-//ftl:verb
-func Empty(ctx context.Context) error {
-	return nil
-}

From 6eba7995b67e9fed67133b6619a765aa5d9aeaf4 Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Wed, 13 Mar 2024 16:03:59 -0700
Subject: [PATCH 5/7] add some integration tests

---
 go-runtime/compile/build.go                   |  3 --
 go.mod                                        |  4 --
 integration/integration_test.go               | 16 ++++++
 integration/testdata/go/pubsub/echo.go        | 49 +++++++++++++++++++
 integration/testdata/kotlin/pubsub/Echo.kt    | 25 ++++++++++
 .../ftl/schemaextractor/ExtractSchemaRule.kt  |  2 +-
 6 files changed, 91 insertions(+), 8 deletions(-)
 create mode 100644 integration/testdata/go/pubsub/echo.go
 create mode 100644 integration/testdata/kotlin/pubsub/Echo.kt

diff --git a/go-runtime/compile/build.go b/go-runtime/compile/build.go
index c64f81103..49da10002 100644
--- a/go-runtime/compile/build.go
+++ b/go-runtime/compile/build.go
@@ -211,9 +211,6 @@ var scaffoldFuncs = scaffolder.FuncMap{
 		}
 		panic(fmt.Sprintf("unsupported value %T", v))
 	},
-	"isSink": func(v schema.Verb) string {
-		return "wes"
-	},
 }
 
 func genType(module *schema.Module, t schema.Type) string {
diff --git a/go.mod b/go.mod
index 1f8f4bdb7..4cf08aa9a 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,6 @@ require (
 	github.com/radovskyb/watcher v1.0.7
 	github.com/rs/cors v1.10.1
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
-	github.com/stretchr/testify v1.8.4
 	github.com/swaggest/jsonschema-go v0.3.69
 	github.com/titanous/json5 v1.0.0
 	github.com/tmc/langchaingo v0.1.5
@@ -58,13 +57,10 @@ require (
 
 require (
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
 	github.com/pkoukk/tiktoken-go v0.1.2 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.1.0 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
 	modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect
 )
 
diff --git a/integration/integration_test.go b/integration/integration_test.go
index 3ce263c6f..248f03f40 100644
--- a/integration/integration_test.go
+++ b/integration/integration_test.go
@@ -213,6 +213,22 @@ func TestExternalCalls(t *testing.T) {
 		return tests
 	})
 }
+func TestPubSub(t *testing.T) {
+	runForRuntimes(t, func(modulesDir string, runtime string, rd runtimeData) []test {
+		return []test{
+			{name: fmt.Sprintf("PubSub%s", rd.testSuffix), assertions: assertions{
+				run("ftl", rd.initOpts...),
+				scaffoldTestData(runtime, "pubsub", rd.modulePath),
+				run("ftl", "deploy", rd.moduleRoot),
+				call(rd.moduleName, "echo", obj{}, func(t testing.TB, resp obj) {
+					name, ok := resp["name"].(string)
+					assert.True(t, ok, "name is not a string")
+					assert.Equal(t, "source", name)
+				}),
+			}},
+		}
+	})
+}
 
 func TestSchemaGenerate(t *testing.T) {
 	tests := []test{
diff --git a/integration/testdata/go/pubsub/echo.go b/integration/testdata/go/pubsub/echo.go
new file mode 100644
index 000000000..d950af883
--- /dev/null
+++ b/integration/testdata/go/pubsub/echo.go
@@ -0,0 +1,49 @@
+//ftl:module echo
+package echo
+
+import (
+	"context"
+
+	"github.com/TBD54566975/ftl/go-runtime/ftl"
+	// Import the FTL SDK.
+)
+
+type EchoRequest struct{}
+type EchoResponse struct {
+	Name string
+}
+
+//ftl:verb
+func Echo(ctx context.Context, req EchoRequest) (EchoResponse, error) {
+	err := ftl.CallSink(ctx, Sink, SinkRequest{})
+	if err != nil {
+		return EchoResponse{}, err
+	}
+	resp, err := ftl.CallSource(ctx, Source)
+	if err != nil {
+		return EchoResponse{}, err
+	}
+
+	name := resp.Name
+	return EchoResponse{
+		Name: name,
+	}, nil
+}
+
+type SourceResponse struct {
+	Name string
+}
+
+//ftl:verb
+func Source(ctx context.Context) (SourceResponse, error) {
+	return SourceResponse{
+		Name: "source",
+	}, nil
+}
+
+type SinkRequest struct{}
+
+//ftl:verb
+func Sink(ctx context.Context, req SinkRequest) error {
+	return nil
+}
diff --git a/integration/testdata/kotlin/pubsub/Echo.kt b/integration/testdata/kotlin/pubsub/Echo.kt
new file mode 100644
index 000000000..be3b16f5a
--- /dev/null
+++ b/integration/testdata/kotlin/pubsub/Echo.kt
@@ -0,0 +1,25 @@
+package ftl.echo
+
+import ftl.builtin.Empty
+import xyz.block.ftl.Context
+import xyz.block.ftl.Verb
+
+data class EchoResponse(val name: String)
+
+@Verb
+fun echo(context: Context, req: Empty): EchoResponse {
+  context.callSink(::sink, Empty())
+  val resp = context.callSource(::source)
+  return EchoResponse(name = resp.name)
+}
+
+data class SourceResponse(val name: String)
+
+@Verb
+fun source(context: Context): EchoResponse {
+  return EchoResponse(name = "source")
+}
+
+@Verb
+fun sink(context: Context, req: Empty) {
+}
diff --git a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
index 9f91df1ec..0a6c9154e 100644
--- a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
+++ b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
@@ -220,7 +220,7 @@ class SchemaExtractor(
       if (verb.valueParameters.size == 2) {
         val reqParam = verb.valueParameters.last()
         require(reqParam.typeReference?.resolveType()
-          ?.let { it.toClassDescriptor().isData || it.isEmptyBuiltin() || it.isUnit() }
+          ?.let { it.toClassDescriptor().isData || it.isEmptyBuiltin() }
           ?: false
         ) {
           "${verb.valueParameters.last().getLineAndColumn()} Second argument of ${verb.name} must be a data class or " +

From cd12047d2789e1bea1cb9b3605e7d23312fa6db6 Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Wed, 13 Mar 2024 17:04:41 -0700
Subject: [PATCH 6/7] pr review changes :)

---
 .../ftl/schemaextractor/ExtractSchemaRule.kt  | 19 ++++++-------------
 1 file changed, 6 insertions(+), 13 deletions(-)

diff --git a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
index 0a6c9154e..79a94f6aa 100644
--- a/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
+++ b/kotlin-runtime/ftl-runtime/src/main/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRule.kt
@@ -242,22 +242,15 @@ class SchemaExtractor(
   private fun extractVerb(verb: KtNamedFunction): Verb {
     val verbSourcePos = verb.getLineAndColumn()
 
-    var requestRef = Type(unit = xyz.block.ftl.v1.schema.Unit())
-    if (verb.valueParameters.size > 1) {
-      val ref = verb.valueParameters.last()?.let {
-        val position = it.getLineAndColumn().toPosition()
-        return@let it.typeReference?.resolveType()?.toSchemaType(position)
-      }
-      ref?.let { requestRef = it }
-
-    }
+    val requestRef = verb.valueParameters.takeIf { it.size > 1 }?.last()?.let {
+      val position = it.getLineAndColumn().toPosition()
+      return@let it.typeReference?.resolveType()?.toSchemaType(position)
+    } ?: Type(unit = xyz.block.ftl.v1.schema.Unit())
 
-    var returnRef = Type(unit = xyz.block.ftl.v1.schema.Unit())
-    val ref = verb.createTypeBindingForReturnType(bindingContext)?.let {
+    val returnRef = verb.createTypeBindingForReturnType(bindingContext)?.let {
       val position = it.psiElement.getLineAndColumn().toPosition()
       return@let it.type.toSchemaType(position)
-    }
-    ref?.let { returnRef = it }
+    } ?: Type(unit = xyz.block.ftl.v1.schema.Unit())
 
     val metadata = mutableListOf<Metadata>()
     extractIngress(verb, requestRef, returnRef)?.apply { metadata.add(Metadata(ingress = this)) }

From 632022e2c1ba1945b7e98aca731e25fb5ee2abd1 Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Thu, 14 Mar 2024 09:18:25 -0700
Subject: [PATCH 7/7] fix up new Ref types

---
 buildengine/build_go_test.go                  |  4 +-
 buildengine/build_kotlin_test.go              |  4 +-
 go-runtime/ftl/call.go                        | 47 +++++--------------
 go-runtime/ftl/types.go                       |  5 +-
 go-runtime/server/server.go                   | 29 ++++--------
 .../schemaextractor/ExtractSchemaRuleTest.kt  |  5 +-
 6 files changed, 32 insertions(+), 62 deletions(-)

diff --git a/buildengine/build_go_test.go b/buildengine/build_go_test.go
index 6b65f6b4a..14b0d610c 100644
--- a/buildengine/build_go_test.go
+++ b/buildengine/build_go_test.go
@@ -39,14 +39,14 @@ func TestGenerateGoModule(t *testing.T) {
 				&schema.Data{Name: "SinkReq"},
 				&schema.Verb{
 					Name:     "sink",
-					Request:  &schema.DataRef{Name: "SinkReq"},
+					Request:  &schema.Ref{Name: "SinkReq"},
 					Response: &schema.Unit{},
 				},
 				&schema.Data{Name: "SourceResp"},
 				&schema.Verb{
 					Name:     "source",
 					Request:  &schema.Unit{},
-					Response: &schema.DataRef{Name: "SourceResp"},
+					Response: &schema.Ref{Name: "SourceResp"},
 				},
 				&schema.Verb{
 					Name:     "nothing",
diff --git a/buildengine/build_kotlin_test.go b/buildengine/build_kotlin_test.go
index 94b0db764..290dc7171 100644
--- a/buildengine/build_kotlin_test.go
+++ b/buildengine/build_kotlin_test.go
@@ -323,7 +323,7 @@ func TestGenerateSourcesAndSinks(t *testing.T) {
 						}},
 					&schema.Verb{
 						Name:     "sink",
-						Request:  &schema.DataRef{Name: "SinkReq"},
+						Request:  &schema.Ref{Name: "SinkReq"},
 						Response: &schema.Unit{},
 					},
 					&schema.Data{
@@ -334,7 +334,7 @@ func TestGenerateSourcesAndSinks(t *testing.T) {
 					&schema.Verb{
 						Name:     "source",
 						Request:  &schema.Unit{},
-						Response: &schema.DataRef{Name: "SourceResp"},
+						Response: &schema.Ref{Name: "SourceResp"},
 					},
 					&schema.Verb{
 						Name:     "nothing",
diff --git a/go-runtime/ftl/call.go b/go-runtime/ftl/call.go
index 9f7f04c74..45b3e35c4 100644
--- a/go-runtime/ftl/call.go
+++ b/go-runtime/ftl/call.go
@@ -17,7 +17,7 @@ import (
 	"github.com/TBD54566975/ftl/internal/rpc"
 )
 
-func call[Req, Resp any](ctx context.Context, callee *schemapb.VerbRef, req Req) (resp Resp, err error) {
+func call[Req, Resp any](ctx context.Context, callee *schemapb.Ref, req Req) (resp Resp, err error) {
 	client := rpc.ClientFromContext[ftlv1connect.VerbServiceClient](ctx)
 	reqData, err := encoding.Marshal(req)
 	if err != nil {
@@ -44,58 +44,37 @@ func call[Req, Resp any](ctx context.Context, callee *schemapb.VerbRef, req Req)
 }
 
 // Call a Verb through the FTL Controller.
-func Call[Req, Resp any](ctx context.Context, verb Verb[Req, Resp], req Req) (resp Resp, err error) {
-	callee := VerbToRef(verb)
-	return call[Req, Resp](ctx, callee.ToProto(), req)
+func Call[Req, Resp any](ctx context.Context, verb Verb[Req, Resp], req Req) (Resp, error) {
+	return call[Req, Resp](ctx, CallToSchemaRef(verb), req)
 }
 
 // CallSink calls a Sink through the FTL controller.
 func CallSink[Req any](ctx context.Context, sink Sink[Req], req Req) error {
-	callee := SinkToRef(sink)
-	verbRef := &schemapb.VerbRef{Module: callee.Module, Name: callee.Name}
-	_, err := call[Req, Unit](ctx, verbRef, req)
+	_, err := call[Req, Unit](ctx, CallToSchemaRef(sink), req)
 	return err
 }
 
 // CallSource calls a Source through the FTL controller.
 func CallSource[Resp any](ctx context.Context, source Source[Resp]) (Resp, error) {
-	callee := SourceToRef(source)
-	verbRef := &schemapb.VerbRef{Module: callee.Module, Name: callee.Name}
-	fmt.Printf("source ref2: %v\n", verbRef)
-	return call[Unit, Resp](ctx, verbRef, Unit{})
+	return call[Unit, Resp](ctx, CallToSchemaRef(source), Unit{})
 }
 
 // CallEmpty calls a Verb with no request or response through the FTL controller.
 func CallEmpty(ctx context.Context, empty Empty) error {
-	callee := EmptyToRef(empty)
-	verbRef := &schemapb.VerbRef{Module: callee.Module, Name: callee.Name}
-
-	_, err := call[Unit, Unit](ctx, verbRef, Unit{})
+	_, err := call[Unit, Unit](ctx, CallToSchemaRef(empty), Unit{})
 	return err
 }
 
-// VerbToRef returns the FTL reference for a Verb.
-func VerbToRef[Req, Resp any](verb Verb[Req, Resp]) Ref {
-	ref := runtime.FuncForPC(reflect.ValueOf(verb).Pointer()).Name()
-	return goRefToFTLRef(ref)
-}
-
-// SinkToRef returns the FTL reference for a Sink.
-func SinkToRef[Req any](sink Sink[Req]) Ref {
-	ref := runtime.FuncForPC(reflect.ValueOf(sink).Pointer()).Name()
+// CallToRef returns the Ref for a Verb, Sink, Source, or Empty.
+func CallToRef(call any) Ref {
+	ref := runtime.FuncForPC(reflect.ValueOf(call).Pointer()).Name()
 	return goRefToFTLRef(ref)
 }
 
-// SourceToRef returns the FTL reference for a Source.
-func SourceToRef[Resp any](source Source[Resp]) Ref {
-	ref := runtime.FuncForPC(reflect.ValueOf(source).Pointer()).Name()
-	return goRefToFTLRef(ref)
-}
-
-// EmptyToRef returns the FTL reference for an Empty.
-func EmptyToRef(empty Empty) Ref {
-	ref := runtime.FuncForPC(reflect.ValueOf(empty).Pointer()).Name()
-	return goRefToFTLRef(ref)
+// CallToSchemaRef returns the Ref for a Verb, Sink, Source, or Empty as a Schema Ref.
+func CallToSchemaRef(call any) *schemapb.Ref {
+	ref := CallToRef(call)
+	return ref.ToProto()
 }
 
 func goRefToFTLRef(ref string) Ref {
diff --git a/go-runtime/ftl/types.go b/go-runtime/ftl/types.go
index c97d63ca3..5af50ff9b 100644
--- a/go-runtime/ftl/types.go
+++ b/go-runtime/ftl/types.go
@@ -52,4 +52,7 @@ type Verb[Req, Resp any] func(context.Context, Req) (Resp, error)
 type Sink[Req any] func(context.Context, Req) error
 
 // A Source is a function that does not accept input but returns output.
-type Source[Req any] func(context.Context, Req) error
+type Source[Resp any] func(context.Context) (Resp, error)
+
+// An Empty is a function that does not accept input or return output.
+type Empty func(context.Context) error
diff --git a/go-runtime/server/server.go b/go-runtime/server/server.go
index ecbd7a3b7..9d1be273b 100644
--- a/go-runtime/server/server.go
+++ b/go-runtime/server/server.go
@@ -65,7 +65,7 @@ type Handler struct {
 	fn  func(ctx context.Context, req []byte) ([]byte, error)
 }
 
-func handler[Req, Resp any](ref ftl.VerbRef, verb func(ctx context.Context, req Req) (Resp, error)) Handler {
+func handler[Req, Resp any](ref ftl.Ref, verb func(ctx context.Context, req Req) (Resp, error)) Handler {
 	return Handler{
 		ref: ref,
 		fn: func(ctx context.Context, reqdata []byte) ([]byte, error) {
@@ -94,43 +94,30 @@ func handler[Req, Resp any](ref ftl.VerbRef, verb func(ctx context.Context, req
 
 // HandleCall creates a Handler from a Verb.
 func HandleCall[Req, Resp any](verb func(ctx context.Context, req Req) (Resp, error)) Handler {
-	return handler(ftl.VerbToRef(verb), verb)
+	return handler(ftl.CallToRef(verb), verb)
 }
 
 // HandleSink creates a Handler from a Sink with no response.
 func HandleSink[Req any](sink func(ctx context.Context, req Req) error) Handler {
-	ref := ftl.SinkToRef(sink)
-
-	verb := func(ctx context.Context, req Req) (ftl.Unit, error) {
+	return handler(ftl.CallToRef(sink), func(ctx context.Context, req Req) (ftl.Unit, error) {
 		err := sink(ctx, req)
 		return ftl.Unit{}, err
-	}
-
-	return handler(ftl.VerbRef{Module: ref.Module, Name: ref.Name}, verb)
+	})
 }
 
 // HandleSource creates a Handler from a Source with no request.
 func HandleSource[Resp any](source func(ctx context.Context) (Resp, error)) Handler {
-	ref := ftl.SourceToRef(source)
-
-	fmt.Printf("source ref: %v\n", ref)
-	verb := func(ctx context.Context, _ ftl.Unit) (Resp, error) {
+	return handler(ftl.CallToRef(source), func(ctx context.Context, _ ftl.Unit) (Resp, error) {
 		return source(ctx)
-	}
-
-	return handler(ftl.VerbRef{Module: ref.Module, Name: ref.Name}, verb)
+	})
 }
 
 // HandleEmpty creates a Handler from a Verb with no request or response.
 func HandleEmpty(empty func(ctx context.Context) error) Handler {
-	ref := ftl.EmptyToRef(empty)
-
-	verb := func(ctx context.Context, _ ftl.Unit) (ftl.Unit, error) {
+	return handler(ftl.CallToRef(empty), func(ctx context.Context, _ ftl.Unit) (ftl.Unit, error) {
 		err := empty(ctx)
 		return ftl.Unit{}, err
-	}
-
-	return handler(ftl.VerbRef{Module: ref.Module, Name: ref.Name}, verb)
+	})
 }
 
 var _ ftlv1connect.VerbServiceHandler = (*moduleServer)(nil)
diff --git a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt
index 167a04fe4..25453fcdf 100644
--- a/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt
+++ b/kotlin-runtime/ftl-runtime/src/test/kotlin/xyz/block/ftl/schemaextractor/ExtractSchemaRuleTest.kt
@@ -34,6 +34,7 @@ import xyz.block.ftl.v1.schema.Ref
 import xyz.block.ftl.v1.schema.StringValue
 import xyz.block.ftl.v1.schema.Type
 import xyz.block.ftl.v1.schema.TypeParameter
+import xyz.block.ftl.v1.schema.Unit
 import xyz.block.ftl.v1.schema.Value
 import xyz.block.ftl.v1.schema.Verb
 
@@ -309,7 +310,7 @@ internal class ExtractSchemaRuleTest(private val env: KotlinCoreEnvironment) {
           verb = Verb(
             name = "sink",
             request = Type(
-              dataRef = DataRef(
+              ref = Ref(
                 name = "Empty",
                 module = "builtin"
               )
@@ -326,7 +327,7 @@ internal class ExtractSchemaRuleTest(private val env: KotlinCoreEnvironment) {
               unit = Unit()
             ),
             response = Type(
-              dataRef = DataRef(
+              ref = Ref(
                 name = "Empty",
                 module = "builtin"
               )