feat: go publishes topic events to kafka (#3577)

First half of [Publish events via runner
#3464](#3464)

Topics take in a partition mapper, with SinglePartitionMap being a
convenience type for when there are no partitions
```go
type AgentBroadcast = ftl.TopicHandle[Agent, ftl.SinglePartitionMap[Agent]]
```

- Go publishes to kafka *and* FTL pubsub. Publishing to ftl pubsub will
be removed soon.
- Integration tests not included yet.

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Justfile

@@ -25,7 +25,8 @@ PROTOS_OUT := "backend/protos/xyz/block/ftl/console/v1/console.pb.go " + \
               "backend/protos/xyz/block/ftl/schema/v1/schema.pb.go " + \
               CONSOLE_ROOT + "/src/protos/xyz/block/ftl/console/v1/console_pb.ts " + \
               CONSOLE_ROOT + "/src/protos/xyz/block/ftl/v1/ftl_pb.ts " + \
-              CONSOLE_ROOT + "/src/protos/xyz/block/ftl/schema/v1/schema_pb.ts"
+              CONSOLE_ROOT + "/src/protos/xyz/block/ftl/schema/v1/schema_pb.ts" + \
+              CONSOLE_ROOT + "/src/protos/xyz/block/ftl/publish/v1/publish_pb.ts"
 # Configuration for building Docker images
 DOCKER_IMAGES := '''
 {

backend/controller/encryption/testdata/go/encryption/encryption.go

@@ -33,7 +33,7 @@ type Event struct {
 //
 // Used to test encryption of topic_events and async_calls tables
 
-type Topic = ftl.TopicHandle[Event]
+type Topic = ftl.TopicHandle[Event, ftl.SinglePartitionMap[Event]]
 
 //ftl:verb
 func Publish(ctx context.Context, e Event, topic Topic) error {

backend/controller/pubsub/testdata/go/publisher/publisher.go

@@ -9,7 +9,7 @@ import (
 )
 
 //ftl:export
-type TestTopic = ftl.TopicHandle[PubSubEvent]
+type TestTopic = ftl.TopicHandle[PubSubEvent, ftl.SinglePartitionMap[PubSubEvent]]
 
 type PubSubEvent struct {
 	Time time.Time
@@ -38,7 +38,7 @@ func PublishOne(ctx context.Context, topic TestTopic) error {
 }
 
 //ftl:export
-type Topic2 = ftl.TopicHandle[PubSubEvent]
+type Topic2 = ftl.TopicHandle[PubSubEvent, ftl.SinglePartitionMap[PubSubEvent]]
 
 //ftl:verb
 func PublishOneToTopic2(ctx context.Context, topic Topic2) error {

backend/controller/pubsub/testdata/go/slow/slow.go

@@ -9,7 +9,7 @@ import (
 	"github.com/TBD54566975/ftl/go-runtime/ftl" // Import the FTL SDK.
 )
 
-type Topic = ftl.TopicHandle[Event]
+type Topic = ftl.TopicHandle[Event, ftl.SinglePartitionMap[Event]]
 
 type Event struct {
 	Duration int

backend/controller/pubsub/testdata/go/subscriber/go.mod

@@ -13,21 +13,16 @@ require (
 	connectrpc.com/grpcreflect v1.2.0 // indirect
 	connectrpc.com/otelconnect v0.7.1 // indirect
 	filippo.io/edwards25519 v1.1.0 // indirect
-	github.com/XSAM/otelsql v0.35.0 // indirect
 	github.com/alecthomas/concurrency v0.0.2 // indirect
-	github.com/alecthomas/kong v1.5.1 // indirect
 	github.com/alecthomas/participle/v2 v2.1.1 // indirect
 	github.com/alecthomas/types v0.17.0 // indirect
 	github.com/benbjohnson/clock v1.3.5 // indirect
-	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/danieljoos/wincred v1.2.2 // indirect
 	github.com/deckarep/golang-set/v2 v2.6.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-sql-driver/mysql v1.8.1 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
-	github.com/google/uuid v1.6.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.23.0 // indirect
 	github.com/hashicorp/cronexpr v1.1.2 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
@@ -42,25 +37,15 @@ require (
 	github.com/swaggest/refl v1.3.0 // indirect
 	github.com/zalando/go-keyring v0.2.6 // indirect
 	go.opentelemetry.io/otel v1.32.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.32.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.32.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.32.0 // indirect
 	go.opentelemetry.io/otel/metric v1.32.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.32.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.32.0 // indirect
 	go.opentelemetry.io/otel/trace v1.32.0 // indirect
-	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
-	go.uber.org/automaxprocs v1.6.0 // indirect
 	golang.org/x/crypto v0.29.0 // indirect
 	golang.org/x/exp v0.0.0-20241108190413-2d47ceb2692f // indirect
 	golang.org/x/mod v0.22.0 // indirect
 	golang.org/x/net v0.31.0 // indirect
 	golang.org/x/sync v0.9.0 // indirect
 	golang.org/x/sys v0.27.0 // indirect
 	golang.org/x/text v0.20.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20241104194629-dd2ea8efbc28 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20241104194629-dd2ea8efbc28 // indirect
-	google.golang.org/grpc v1.68.0 // indirect
 	google.golang.org/protobuf v1.35.2 // indirect
 )
 

backend/controller/pubsub/testdata/go/subscriber/subscriber.go

@@ -40,7 +40,7 @@ func ConsumeButFailAndCatchAny(ctx context.Context, req publisher.PubSubEvent) e
 //ftl:verb
 func PublishToExternalModule(ctx context.Context) error {
 	// Get around compile-time checks
-	externalTopic := ftl.TopicHandle[publisher.PubSubEvent]{Ref: reflection.Ref{Module: "publisher", Name: "testTopic"}.ToSchema()}
+	externalTopic := ftl.TopicHandle[publisher.PubSubEvent, ftl.SinglePartitionMap[publisher.PubSubEvent]]{Ref: reflection.Ref{Module: "publisher", Name: "testTopic"}.ToSchema()}
 	return externalTopic.Publish(ctx, publisher.PubSubEvent{Time: time.Now()})
 }