chore: move timeline client into internal

Seperate the service implementation from the client API.
block · Dec 24, 2024 · 20ad3ab · 20ad3ab
1 parent b9dde97
commit 20ad3ab
Show file tree

Hide file tree

Showing 33 changed files with 133 additions and 137 deletions.
diff --git a/backend/console/console.go b/backend/console/console.go
@@ -15,7 +15,6 @@ import (
 	timelinepb "github.com/block/ftl/backend/protos/xyz/block/ftl/timeline/v1"
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
 	"github.com/block/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
-	"github.com/block/ftl/backend/timeline"
 	schemapb "github.com/block/ftl/common/protos/xyz/block/ftl/schema/v1"
 	"github.com/block/ftl/common/schema"
 	frontend "github.com/block/ftl/frontend/console"
@@ -25,6 +24,7 @@ import (
 	"github.com/block/ftl/internal/routing"
 	"github.com/block/ftl/internal/rpc"
 	"github.com/block/ftl/internal/schema/schemaeventsource"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 type Config struct {
@@ -36,14 +36,14 @@ type Config struct {
 type service struct {
 	schemaEventSource schemaeventsource.EventSource
 	controllerClient  ftlv1connect.ControllerServiceClient
-	timelineClient    *timeline.Client
+	timelineClient    *timelineclient.Client
 	adminClient       admin.Client
 	callClient        routing.CallClient
 }
 
 var _ consolepbconnect.ConsoleServiceHandler = (*service)(nil)
 
-func Start(ctx context.Context, config Config, eventSource schemaeventsource.EventSource, controllerClient ftlv1connect.ControllerServiceClient, timelineClient *timeline.Client, adminClient admin.Client, client routing.CallClient) error {
+func Start(ctx context.Context, config Config, eventSource schemaeventsource.EventSource, controllerClient ftlv1connect.ControllerServiceClient, timelineClient *timelineclient.Client, adminClient admin.Client, client routing.CallClient) error {
 	logger := log.FromContext(ctx).Scope("console")
 	ctx = log.ContextWithLogger(ctx, logger)
 

diff --git a/backend/controller/controller.go b/backend/controller/controller.go
@@ -37,7 +37,6 @@ import (
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
 	"github.com/block/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
 	"github.com/block/ftl/backend/runner/pubsub"
-	"github.com/block/ftl/backend/timeline"
 	schemapb "github.com/block/ftl/common/protos/xyz/block/ftl/schema/v1"
 	"github.com/block/ftl/common/schema"
 	"github.com/block/ftl/common/sha256"
@@ -52,6 +51,7 @@ import (
 	"github.com/block/ftl/internal/rpc"
 	"github.com/block/ftl/internal/rpc/headers"
 	"github.com/block/ftl/internal/schema/schemaeventsource"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 // CommonConfig between the production controller and development server.
@@ -95,7 +95,7 @@ func Start(
 	config Config,
 	storage *artefacts.OCIArtefactService,
 	adminClient ftlv1connect.AdminServiceClient,
-	timelineClient *timeline.Client,
+	timelineClient *timelineclient.Client,
 	devel bool,
 ) error {
 	config.SetDefaults()
@@ -140,7 +140,7 @@ type Service struct {
 
 	tasks          *scheduledtask.Scheduler
 	pubSub         *pubsub.Service
-	timelineClient *timeline.Client
+	timelineClient *timelineclient.Client
 	storage        *artefacts.OCIArtefactService
 
 	// Map from runnerKey.String() to client.
@@ -156,7 +156,7 @@ type Service struct {
 func New(
 	ctx context.Context,
 	adminClient ftlv1connect.AdminServiceClient,
-	timelineClient *timeline.Client,
+	timelineClient *timelineclient.Client,
 	storage *artefacts.OCIArtefactService,
 	config Config,
 	devel bool,
@@ -418,7 +418,7 @@ func (s *Service) setDeploymentReplicas(ctx context.Context, key model.Deploymen
 			return fmt.Errorf("could not activate deployment: %w", err)
 		}
 	}
-	s.timelineClient.Publish(ctx, timeline.DeploymentUpdated{
+	s.timelineClient.Publish(ctx, timelineclient.DeploymentUpdated{
 		DeploymentKey:   key,
 		MinReplicas:     minReplicas,
 		PrevMinReplicas: deployment.MinReplicas,
@@ -481,7 +481,7 @@ func (s *Service) ReplaceDeploy(ctx context.Context, c *connect.Request[ftlv1.Re
 		}
 	}
 
-	s.timelineClient.Publish(ctx, timeline.DeploymentCreated{
+	s.timelineClient.Publish(ctx, timelineclient.DeploymentCreated{
 		DeploymentKey:      newDeploymentKey,
 		Language:           newDeployment.Language,
 		ModuleName:         newDeployment.Module,
@@ -842,7 +842,7 @@ func (s *Service) callWithRequest(
 		return nil, fmt.Errorf("deployment not found for module %q", module)
 	}
 
-	callEvent := &timeline.Call{
+	callEvent := &timelineclient.Call{
 		DeploymentKey:    deployment,
 		RequestKey:       requestKey,
 		ParentRequestKey: parentKey,

diff --git a/backend/controller/deployment_logs.go b/backend/controller/deployment_logs.go
@@ -6,15 +6,15 @@ import (
 
 	"github.com/alecthomas/types/optional"
 
-	"github.com/block/ftl/backend/timeline"
 	"github.com/block/ftl/internal/channels"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/model"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 var _ log.Sink = (*deploymentLogsSink)(nil)
 
-func newDeploymentLogsSink(ctx context.Context, timelineClient *timeline.Client) *deploymentLogsSink {
+func newDeploymentLogsSink(ctx context.Context, timelineClient *timelineclient.Client) *deploymentLogsSink {
 	sink := &deploymentLogsSink{
 		logQueue: make(chan log.Entry, 10000),
 	}
@@ -40,7 +40,7 @@ func (d *deploymentLogsSink) Log(entry log.Entry) error {
 	return nil
 }
 
-func (d *deploymentLogsSink) processLogs(ctx context.Context, timelineClient *timeline.Client) {
+func (d *deploymentLogsSink) processLogs(ctx context.Context, timelineClient *timelineclient.Client) {
 	for entry := range channels.IterContext(ctx, d.logQueue) {
 		var deployment model.DeploymentKey
 		depStr, ok := entry.Attributes["deployment"]
@@ -67,7 +67,7 @@ func (d *deploymentLogsSink) processLogs(ctx context.Context, timelineClient *ti
 			errorStr = optional.Some(entry.Error.Error())
 		}
 
-		timelineClient.Publish(ctx, &timeline.Log{
+		timelineClient.Publish(ctx, &timelineclient.Log{
 			RequestKey:    request,
 			DeploymentKey: deployment,
 			Time:          entry.Time,

diff --git a/backend/cron/service.go b/backend/cron/service.go
@@ -11,7 +11,6 @@ import (
 
 	"github.com/block/ftl/backend/cron/observability"
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
-	"github.com/block/ftl/backend/timeline"
 	"github.com/block/ftl/common/cron"
 	"github.com/block/ftl/common/schema"
 	"github.com/block/ftl/common/slices"
@@ -20,6 +19,7 @@ import (
 	"github.com/block/ftl/internal/routing"
 	"github.com/block/ftl/internal/rpc/headers"
 	"github.com/block/ftl/internal/schema/schemaeventsource"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 type cronJob struct {
@@ -46,7 +46,7 @@ func (c cronJob) String() string {
 }
 
 // Start the cron service. Blocks until the context is cancelled.
-func Start(ctx context.Context, eventSource schemaeventsource.EventSource, client routing.CallClient, timelineClient *timeline.Client) error {
+func Start(ctx context.Context, eventSource schemaeventsource.EventSource, client routing.CallClient, timelineClient *timelineclient.Client) error {
 	logger := log.FromContext(ctx).Scope("cron")
 	ctx = log.ContextWithLogger(ctx, logger)
 	// Map of cron jobs for each module.
@@ -136,11 +136,11 @@ func callCronJob(ctx context.Context, verbClient routing.CallClient, cronJob cro
 	}
 }
 
-func scheduleNext(ctx context.Context, cronQueue []cronJob, timelineClient *timeline.Client) (time.Duration, bool) {
+func scheduleNext(ctx context.Context, cronQueue []cronJob, timelineClient *timelineclient.Client) (time.Duration, bool) {
 	if len(cronQueue) == 0 {
 		return 0, false
 	}
-	timelineClient.Publish(ctx, timeline.CronScheduled{
+	timelineClient.Publish(ctx, timelineclient.CronScheduled{
 		DeploymentKey: cronQueue[0].deployment,
 		Verb:          schema.Ref{Module: cronQueue[0].module, Name: cronQueue[0].verb.Name},
 		ScheduledAt:   cronQueue[0].next,

diff --git a/backend/cron/service_test.go b/backend/cron/service_test.go
@@ -15,13 +15,13 @@ import (
 	"github.com/alecthomas/types/optional"
 
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
-	"github.com/block/ftl/backend/timeline"
 	schemapb "github.com/block/ftl/common/protos/xyz/block/ftl/schema/v1"
 	"github.com/block/ftl/common/schema"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/model"
 	"github.com/block/ftl/internal/routing"
 	"github.com/block/ftl/internal/schema/schemaeventsource"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 type verbClient struct {
@@ -72,7 +72,7 @@ func TestCron(t *testing.T) {
 	timelineEndpoint, err := url.Parse("http://localhost:8080")
 	assert.NoError(t, err)
 
-	timelineClient := timeline.NewClient(ctx, timelineEndpoint)
+	timelineClient := timelineclient.NewClient(ctx, timelineEndpoint)
 	ctx, cancel := context.WithTimeout(ctx, time.Second*5)
 	t.Cleanup(cancel)
 

diff --git a/backend/ingress/handler.go b/backend/ingress/handler.go
@@ -12,12 +12,12 @@ import (
 	"github.com/alecthomas/types/optional"
 
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
-	"github.com/block/ftl/backend/timeline"
 	schemapb "github.com/block/ftl/common/protos/xyz/block/ftl/schema/v1"
 	"github.com/block/ftl/common/schema"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/model"
 	"github.com/block/ftl/internal/routing"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 // handleHTTP HTTP ingress routes.
@@ -37,7 +37,7 @@ func (s *service) handleHTTP(startTime time.Time, sch *schema.Schema, requestKey
 
 	verbRef := &schemapb.Ref{Module: route.module, Name: route.verb}
 
-	ingressEvent := timeline.Ingress{
+	ingressEvent := timelineclient.Ingress{
 		RequestKey:      requestKey,
 		StartTime:       startTime,
 		Verb:            &schema.Ref{Name: route.verb, Module: route.module},
@@ -154,7 +154,7 @@ func (s *service) handleHTTP(startTime time.Time, sch *schema.Schema, requestKey
 
 func (s *service) recordIngressErrorEvent(
 	ctx context.Context,
-	ingressEvent timeline.Ingress,
+	ingressEvent timelineclient.Ingress,
 	statusCode int,
 	errorMsg string,
 ) {

diff --git a/backend/ingress/handler_test.go b/backend/ingress/handler_test.go
@@ -14,12 +14,12 @@ import (
 	"github.com/alecthomas/types/optional"
 
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
-	"github.com/block/ftl/backend/timeline"
 	"github.com/block/ftl/common/encoding"
 	"github.com/block/ftl/common/schema"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/model"
 	"github.com/block/ftl/internal/schema/schemaeventsource"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 func TestIngress(t *testing.T) {
@@ -108,7 +108,7 @@ func TestIngress(t *testing.T) {
 			svc := &service{
 				view:           syncView(ctx, schemaeventsource.NewUnattached()),
 				client:         fv,
-				timelineClient: timeline.NewClient(ctx, timelineEndpoint),
+				timelineClient: timelineclient.NewClient(ctx, timelineEndpoint),
 			}
 			svc.handleHTTP(time.Now(), sch, reqKey, routes, rec, req, fv)
 			result := rec.Result()

diff --git a/backend/ingress/service.go b/backend/ingress/service.go
@@ -12,7 +12,6 @@ import (
 	"github.com/alecthomas/types/optional"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 
-	"github.com/block/ftl/backend/timeline"
 	schemapb "github.com/block/ftl/common/protos/xyz/block/ftl/schema/v1"
 	"github.com/block/ftl/common/slices"
 	"github.com/block/ftl/internal/cors"
@@ -21,6 +20,7 @@ import (
 	"github.com/block/ftl/internal/model"
 	"github.com/block/ftl/internal/routing"
 	"github.com/block/ftl/internal/schema/schemaeventsource"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 type Config struct {
@@ -40,11 +40,11 @@ type service struct {
 	// Complete schema synchronised from the database.
 	view           *atomic.Value[materialisedView]
 	client         routing.CallClient
-	timelineClient *timeline.Client
+	timelineClient *timelineclient.Client
 }
 
 // Start the HTTP ingress service. Blocks until the context is cancelled.
-func Start(ctx context.Context, config Config, schemaEventSource schemaeventsource.EventSource, client routing.CallClient, timelineClient *timeline.Client) error {
+func Start(ctx context.Context, config Config, schemaEventSource schemaeventsource.EventSource, client routing.CallClient, timelineClient *timelineclient.Client) error {
 	logger := log.FromContext(ctx).Scope("http-ingress")
 	ctx = log.ContextWithLogger(ctx, logger)
 	svc := &service{

diff --git a/backend/runner/proxy/proxy.go b/backend/runner/proxy/proxy.go
@@ -17,12 +17,12 @@ import (
 	ftlleaseconnect "github.com/block/ftl/backend/protos/xyz/block/ftl/lease/v1/leasepbconnect"
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
 	"github.com/block/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
-	"github.com/block/ftl/backend/timeline"
 	"github.com/block/ftl/common/schema"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/model"
 	"github.com/block/ftl/internal/rpc"
 	"github.com/block/ftl/internal/rpc/headers"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 var _ ftlv1connect.VerbServiceHandler = &Service{}
@@ -37,10 +37,10 @@ type Service struct {
 	controllerDeploymentService ftldeploymentconnect.DeploymentServiceClient
 	controllerLeaseService      ftlleaseconnect.LeaseServiceClient
 	moduleVerbService           *xsync.MapOf[string, moduleVerbService]
-	timelineClient              *timeline.Client
+	timelineClient              *timelineclient.Client
 }
 
-func New(controllerModuleService ftldeploymentconnect.DeploymentServiceClient, leaseClient ftlleaseconnect.LeaseServiceClient, timelineClient *timeline.Client) *Service {
+func New(controllerModuleService ftldeploymentconnect.DeploymentServiceClient, leaseClient ftlleaseconnect.LeaseServiceClient, timelineClient *timelineclient.Client) *Service {
 	proxy := &Service{
 		controllerDeploymentService: controllerModuleService,
 		controllerLeaseService:      leaseClient,
@@ -142,7 +142,7 @@ func (r *Service) Call(ctx context.Context, req *connect.Request[ftlv1.CallReque
 		headers.SetRequestKey(req.Header(), requestKey)
 	}
 
-	callEvent := &timeline.Call{
+	callEvent := &timelineclient.Call{
 		DeploymentKey: verbService.deployment,
 		RequestKey:    requestKey,
 		StartTime:     start,

diff --git a/backend/runner/pubsub/consumer.go b/backend/runner/pubsub/consumer.go
@@ -16,13 +16,13 @@ import (
 
 	"github.com/block/ftl/backend/controller/observability"
 	ftlv1 "github.com/block/ftl/backend/protos/xyz/block/ftl/v1"
-	"github.com/block/ftl/backend/timeline"
 	"github.com/block/ftl/common/encoding"
 	"github.com/block/ftl/common/schema"
 	"github.com/block/ftl/common/slices"
 	"github.com/block/ftl/internal/channels"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/model"
+	"github.com/block/ftl/internal/timelineclient"
 )
 
 type consumer struct {
@@ -35,11 +35,11 @@ type consumer struct {
 	deadLetterPublisher optional.Option[*publisher]
 
 	verbClient     VerbClient
-	timelineClient *timeline.Client
+	timelineClient *timelineclient.Client
 }
 
 func newConsumer(moduleName string, verb *schema.Verb, subscriber *schema.MetadataSubscriber, deployment model.DeploymentKey,
-	deadLetterPublisher optional.Option[*publisher], verbClient VerbClient, timelineClient *timeline.Client) (*consumer, error) {
+	deadLetterPublisher optional.Option[*publisher], verbClient VerbClient, timelineClient *timelineclient.Client) (*consumer, error) {
 	if verb.Runtime == nil {
 		return nil, fmt.Errorf("subscription %s has no runtime", verb.Name)
 	}
@@ -211,7 +211,7 @@ func (c *consumer) call(ctx context.Context, body []byte, partition, offset int)
 		Verb: schema.RefKey{Module: c.moduleName, Name: c.verb.Name}.ToProto(),
 		Body: body,
 	}
-	consumeEvent := timeline.PubSubConsume{
+	consumeEvent := timelineclient.PubSubConsume{
 		DeploymentKey: c.deployment,
 		RequestKey:    optional.Some(requestKey.String()),
 		Time:          time.Now(),
@@ -222,7 +222,7 @@ func (c *consumer) call(ctx context.Context, body []byte, partition, offset int)
 	}
 	defer c.timelineClient.Publish(ctx, consumeEvent)
 
-	callEvent := &timeline.Call{
+	callEvent := &timelineclient.Call{
 		DeploymentKey: c.deployment,
 		RequestKey:    requestKey,
 		StartTime:     start,