refactor: use generics to eliminate rudundant logic

Author: kmendell

.github/.golangci.yml

@@ -1,4 +1,4 @@
-version: "2"
+version: '2'
 run:
   tests: false
   timeout: 5m
@@ -11,7 +11,6 @@ linters:
     - asciicheck
     - bidichk
     - bodyclose
-    - canonicalheader
     - clickhouselint
     - contextcheck
     - copyloopvar
@@ -114,13 +113,13 @@ linters:
     rules:
       - linters:
           - revive
-        text: "var-naming:"
+        text: 'var-naming:'
       - linters:
           - revive
-        text: "unused-parameter:"
+        text: 'unused-parameter:'
       - linters:
           - revive
-        text: "exported:"
+        text: 'exported:'
       - linters:
           - forbidigo
         path: cli/

.vscode/settings.json

@@ -22,5 +22,16 @@
   },
   "tasks.statusbar.default.hide": true,
   "tasks.statusbar.limit": 8,
-  "js/ts.tsdk.path": "node_modules/@typescript/native-preview"
+  "js/ts.tsdk.path": "node_modules/@typescript/native-preview",
+  "yaml.disableSchemaDetection": [
+    "**/.github/workflows/*.yml",
+    "**/.github/workflows/*.yaml",
+    "**/.gitea/workflows/*.yml",
+    "**/.gitea/workflows/*.yaml",
+    "**/.forgejo/workflows/*.yml",
+    "**/.forgejo/workflows/*.yaml"
+  ],
+  "yaml.schemas": {
+    "https://json.schemastore.org/github-workflow.json": ".github/workflows/*"
+  }
 }

backend/internal/activity/handler.go

@@ -56,28 +56,16 @@ type ListActivitiesInput struct {
 	ResourceType  string `query:"resourceType" doc:"Filter by resource type"`
 }
 
-type ListActivitiesOutput struct {
-	Body base.Paginated[activitytypes.Activity]
-}
-
 type GetActivityInput struct {
 	EnvironmentID string `path:"id" doc:"Environment ID"`
 	ActivityID    string `path:"activityId" doc:"Activity ID"`
 	Limit         int    `query:"limit" default:"500" doc:"Maximum messages to return"`
 }
 
-type GetActivityOutput struct {
-	Body base.ApiResponse[activitytypes.Detail]
-}
-
 type ClearActivityHistoryInput struct {
 	EnvironmentID string `path:"id" doc:"Environment ID"`
 }
 
-type ClearActivityHistoryOutput struct {
-	Body base.ApiResponse[activitytypes.ClearHistoryResult]
-}
-
 type StreamAllActivitiesInput struct {
 	Limit int `query:"limit" default:"50" doc:"Snapshot limit per environment"`
 }
@@ -95,10 +83,6 @@ type CancelActivityInput struct {
 	RequestedBy   string `query:"requestedBy" doc:"Display name to attribute the cancellation to (used when proxying to a remote environment)"`
 }
 
-type CancelActivityOutput struct {
-	Body base.ApiResponse[activitytypes.Activity]
-}
-
 func NewHandler(activityService *ActivityService, environment EnvironmentDependencies) *ActivityHandler {
 	return &ActivityHandler{
 		activityService: activityService,
@@ -149,7 +133,7 @@ func RegisterActivities(api huma.API, h *ActivityHandler) {
 	}, authz.PermActivitiesDelete, h.ClearHistory)
 }
 
-func (h *ActivityHandler) ListActivities(ctx context.Context, input *ListActivitiesInput) (*ListActivitiesOutput, error) {
+func (h *ActivityHandler) ListActivities(ctx context.Context, input *ListActivitiesInput) (*handlerutil.Page[activitytypes.Activity], error) {
 	if input.EnvironmentID != "0" {
 		return h.proxyListActivitiesInternal(ctx, input)
 	}
@@ -171,7 +155,7 @@ func (h *ActivityHandler) ListActivities(ctx context.Context, input *ListActivit
 	}
 	h.applyActivitySourceLabelsInternal(ctx, input.EnvironmentID, activities)
 
-	return &ListActivitiesOutput{
+	return &handlerutil.Page[activitytypes.Activity]{
 		Body: base.Paginated[activitytypes.Activity]{
 			Success:    true,
 			Data:       activities,
@@ -180,7 +164,7 @@ func (h *ActivityHandler) ListActivities(ctx context.Context, input *ListActivit
 	}, nil
 }
 
-func (h *ActivityHandler) GetActivity(ctx context.Context, input *GetActivityInput) (*GetActivityOutput, error) {
+func (h *ActivityHandler) GetActivity(ctx context.Context, input *GetActivityInput) (*handlerutil.Out[activitytypes.Detail], error) {
 	if input.EnvironmentID != "0" {
 		return h.proxyGetActivityInternal(ctx, input)
 	}
@@ -197,15 +181,15 @@ func (h *ActivityHandler) GetActivity(ctx context.Context, input *GetActivityInp
 	}
 	h.applyActivitySourceLabelInternal(ctx, input.EnvironmentID, &detail.Activity)
 
-	return &GetActivityOutput{
+	return &handlerutil.Out[activitytypes.Detail]{
 		Body: base.ApiResponse[activitytypes.Detail]{
 			Success: true,
 			Data:    *detail,
 		},
 	}, nil
 }
 
-func (h *ActivityHandler) ClearHistory(ctx context.Context, input *ClearActivityHistoryInput) (*ClearActivityHistoryOutput, error) {
+func (h *ActivityHandler) ClearHistory(ctx context.Context, input *ClearActivityHistoryInput) (*handlerutil.Out[activitytypes.ClearHistoryResult], error) {
 	if input.EnvironmentID != "0" {
 		return h.proxyClearHistoryInternal(ctx, input)
 	}
@@ -215,15 +199,15 @@ func (h *ActivityHandler) ClearHistory(ctx context.Context, input *ClearActivity
 		return nil, huma.Error500InternalServerError(err.Error())
 	}
 
-	return &ClearActivityHistoryOutput{
+	return &handlerutil.Out[activitytypes.ClearHistoryResult]{
 		Body: base.ApiResponse[activitytypes.ClearHistoryResult]{
 			Success: true,
 			Data:    activitytypes.ClearHistoryResult{Deleted: deleted},
 		},
 	}, nil
 }
 
-func (h *ActivityHandler) CancelActivity(ctx context.Context, input *CancelActivityInput) (*CancelActivityOutput, error) {
+func (h *ActivityHandler) CancelActivity(ctx context.Context, input *CancelActivityInput) (*handlerutil.Out[activitytypes.Activity], error) {
 	if input.EnvironmentID != "0" {
 		return h.proxyCancelActivityInternal(ctx, input)
 	}
@@ -245,25 +229,25 @@ func (h *ActivityHandler) CancelActivity(ctx context.Context, input *CancelActiv
 	}
 	h.applyActivitySourceLabelInternal(ctx, input.EnvironmentID, cancelled)
 
-	return &CancelActivityOutput{
+	return &handlerutil.Out[activitytypes.Activity]{
 		Body: base.ApiResponse[activitytypes.Activity]{
 			Success: true,
 			Data:    *cancelled,
 		},
 	}, nil
 }
 
-func (h *ActivityHandler) proxyCancelActivityInternal(ctx context.Context, input *CancelActivityInput) (*CancelActivityOutput, error) {
+func (h *ActivityHandler) proxyCancelActivityInternal(ctx context.Context, input *CancelActivityInput) (*handlerutil.Out[activitytypes.Activity], error) {
 	path := fmt.Sprintf("/api/environments/0/activities/%s/cancel", url.PathEscape(input.ActivityID))
 	if requestedBy := h.cancelRequestedByInternal(ctx, input.RequestedBy); requestedBy != "" {
 		path += "?requestedBy=" + url.QueryEscape(requestedBy)
 	}
-	out, err := handlerutil.ProxyRemoteJSON[base.ApiResponse[activitytypes.Activity]](ctx, h.environment.ProxyJSONRequest, input.EnvironmentID, http.MethodPost, path, nil)
+	out, err := h.environment.ProxyJSONRequest.JSON[base.ApiResponse[activitytypes.Activity]](ctx, input.EnvironmentID, http.MethodPost, path, nil)
 	if err != nil {
 		return nil, err
 	}
 	h.applyActivitySourceLabelInternal(ctx, input.EnvironmentID, &out.Data)
-	return &CancelActivityOutput{Body: *out}, nil
+	return &handlerutil.Out[activitytypes.Activity]{Body: *out}, nil
 }
 
 // cancelRequestedByInternal resolves a human-readable name for the cancellation
@@ -480,42 +464,42 @@ func (h *ActivityHandler) runRemoteActivityStreamPollerInternal(ctx context.Cont
 	}
 }
 
-func (h *ActivityHandler) proxyListActivitiesInternal(ctx context.Context, input *ListActivitiesInput) (*ListActivitiesOutput, error) {
+func (h *ActivityHandler) proxyListActivitiesInternal(ctx context.Context, input *ListActivitiesInput) (*handlerutil.Page[activitytypes.Activity], error) {
 	path := "/api/environments/0/activities?" + activityListQueryInternal(input).Encode()
-	out, err := handlerutil.ProxyRemoteJSON[base.Paginated[activitytypes.Activity]](ctx, h.environment.ProxyJSONRequest, input.EnvironmentID, http.MethodGet, path, nil)
+	out, err := h.environment.ProxyJSONRequest.JSON[base.Paginated[activitytypes.Activity]](ctx, input.EnvironmentID, http.MethodGet, path, nil)
 	if err != nil {
 		return nil, err
 	}
 	h.applyActivitySourceLabelsInternal(ctx, input.EnvironmentID, out.Data)
-	return &ListActivitiesOutput{Body: *out}, nil
+	return &handlerutil.Page[activitytypes.Activity]{Body: *out}, nil
 }
 
-func (h *ActivityHandler) proxyListActivitiesForEnvironmentInternal(ctx context.Context, environment environment.Environment, input *ListActivitiesInput) (*ListActivitiesOutput, error) {
+func (h *ActivityHandler) proxyListActivitiesForEnvironmentInternal(ctx context.Context, environment environment.Environment, input *ListActivitiesInput) (*handlerutil.Page[activitytypes.Activity], error) {
 	path := "/api/environments/0/activities?" + activityListQueryInternal(input).Encode()
 	var out base.Paginated[activitytypes.Activity]
 	if err := h.environment.ProxyJSONRequestForEnvironment(ctx, environment, http.MethodGet, path, nil, &out); err != nil {
 		return nil, handlerutil.TranslateRemoteProxyError(err)
 	}
 	applyActivitySourceLabelsForEnvironmentInternal(environment, out.Data)
-	return &ListActivitiesOutput{Body: out}, nil
+	return &handlerutil.Page[activitytypes.Activity]{Body: out}, nil
 }
 
-func (h *ActivityHandler) proxyGetActivityInternal(ctx context.Context, input *GetActivityInput) (*GetActivityOutput, error) {
+func (h *ActivityHandler) proxyGetActivityInternal(ctx context.Context, input *GetActivityInput) (*handlerutil.Out[activitytypes.Detail], error) {
 	path := fmt.Sprintf("/api/environments/0/activities/%s?limit=%d", url.PathEscape(input.ActivityID), input.Limit)
-	out, err := handlerutil.ProxyRemoteJSON[base.ApiResponse[activitytypes.Detail]](ctx, h.environment.ProxyJSONRequest, input.EnvironmentID, http.MethodGet, path, nil)
+	out, err := h.environment.ProxyJSONRequest.JSON[base.ApiResponse[activitytypes.Detail]](ctx, input.EnvironmentID, http.MethodGet, path, nil)
 	if err != nil {
 		return nil, err
 	}
 	h.applyActivitySourceLabelInternal(ctx, input.EnvironmentID, &out.Data.Activity)
-	return &GetActivityOutput{Body: *out}, nil
+	return &handlerutil.Out[activitytypes.Detail]{Body: *out}, nil
 }
 
-func (h *ActivityHandler) proxyClearHistoryInternal(ctx context.Context, input *ClearActivityHistoryInput) (*ClearActivityHistoryOutput, error) {
-	out, err := handlerutil.ProxyRemoteJSON[base.ApiResponse[activitytypes.ClearHistoryResult]](ctx, h.environment.ProxyJSONRequest, input.EnvironmentID, http.MethodDelete, "/api/environments/0/activities/history", nil)
+func (h *ActivityHandler) proxyClearHistoryInternal(ctx context.Context, input *ClearActivityHistoryInput) (*handlerutil.Out[activitytypes.ClearHistoryResult], error) {
+	out, err := h.environment.ProxyJSONRequest.JSON[base.ApiResponse[activitytypes.ClearHistoryResult]](ctx, input.EnvironmentID, http.MethodDelete, "/api/environments/0/activities/history", nil)
 	if err != nil {
 		return nil, err
 	}
-	return &ClearActivityHistoryOutput{Body: *out}, nil
+	return &handlerutil.Out[activitytypes.ClearHistoryResult]{Body: *out}, nil
 }
 
 func (h *ActivityHandler) applyActivitySourceLabelsInternal(ctx context.Context, environmentID string, activities []activitytypes.Activity) {

backend/internal/actors/actor.go

@@ -113,12 +113,12 @@ func (a *Actor) Stop(ctx context.Context) error {
 }
 
 // Request sends one typed request and consumes its response.
-func Request[K comparable, Q, R any](ctx context.Context, a *Actor, request Message[K, Q]) (Message[K, R], error) {
+func (a *Actor) Request[K comparable, Q, R any](ctx context.Context, request Message[K, Q]) (Message[K, R], error) {
 	if a == nil {
 		var zero Message[K, R]
 		return zero, errors.New("actor unavailable")
 	}
-	return requestHandleInternal[K, Q, R](ctx, a.handle, request)
+	return a.handle.request[K, Q, R](ctx, request)
 }
 
 func (a *behaviorActorInternal) Receive(ctx *actor.Context) {

backend/internal/actors/executor.go

@@ -65,9 +65,9 @@ func NewExecutor(ctx context.Context, runtime *Runtime, kind, id string, maxRest
 
 // Execute queues work, waits for its typed result, and runs after only once the
 // result is available to the caller. Work and after are both panic-contained.
-func Execute[T any](ctx context.Context, executor *Executor, label string, work func(context.Context) (T, error), after func(T, error)) (T, error) {
+func (e *Executor) Execute[T any](ctx context.Context, label string, work func(context.Context) (T, error), after func(T, error)) (T, error) {
 	var zero T
-	task, err := Submit(ctx, executor, label, work, after)
+	task, err := e.Submit(ctx, label, work, after)
 	if err != nil {
 		return zero, err
 	}
@@ -77,8 +77,8 @@ func Execute[T any](ctx context.Context, executor *Executor, label string, work
 // Submit queues work synchronously and returns a handle that can be awaited
 // separately. This is useful for terminal cleanup that must stay queued even
 // when the shutdown wait context expires.
-func Submit[T any](ctx context.Context, executor *Executor, label string, work func(context.Context) (T, error), after func(T, error)) (*Task[T], error) {
-	if executor == nil || executor.actor == nil {
+func (e *Executor) Submit[T any](ctx context.Context, label string, work func(context.Context) (T, error), after func(T, error)) (*Task[T], error) {
+	if e == nil || e.actor == nil {
 		return nil, errors.New("actor executor unavailable")
 	}
 	if ctx == nil {
@@ -92,7 +92,7 @@ func Submit[T any](ctx context.Context, executor *Executor, label string, work f
 	}
 
 	result := NewPromise[executorResultInternal[T]]()
-	if err := executor.actor.Send(executorTaskValueInternal[T]{
+	if err := e.actor.Send(executorTaskValueInternal[T]{
 		ctx:     ctx,
 		label:   label,
 		work:    work,
@@ -102,7 +102,7 @@ func Submit[T any](ctx context.Context, executor *Executor, label string, work f
 	}); err != nil {
 		return nil, err
 	}
-	return &Task[T]{result: result, actor: executor.actor}, nil
+	return &Task[T]{result: result, actor: e.actor}, nil
 }
 
 // Wait waits for the submitted task result without changing the task lifetime.
@@ -114,7 +114,7 @@ func (t *Task[T]) Wait(ctx context.Context) (T, error) {
 	if ctx == nil {
 		return zero, errors.New("task wait context unavailable")
 	}
-	completed, err := awaitInternal(ctx, t.actor.handle, t.result, errors.New("actor executor stopped"))
+	completed, err := t.actor.handle.await(ctx, t.result, errors.New("actor executor stopped"))
 	if err != nil {
 		return zero, err
 	}

backend/internal/actors/executor_test.go

@@ -20,7 +20,7 @@ func TestExecutorSerializesHeterogeneousTasksInternal(t *testing.T) {
 	releaseFirst := make(chan struct{})
 	firstResult := make(chan error, 1)
 	go func() {
-		_, executeErr := Execute(t.Context(), executor, "first", func(context.Context) (int, error) {
+		_, executeErr := executor.Execute(t.Context(), "first", func(context.Context) (int, error) {
 			close(firstStarted)
 			<-releaseFirst
 			return 1, nil
@@ -32,7 +32,7 @@ func TestExecutorSerializesHeterogeneousTasksInternal(t *testing.T) {
 	secondStarted := make(chan struct{})
 	secondResult := make(chan string, 1)
 	go func() {
-		value, executeErr := Execute(t.Context(), executor, "second", func(context.Context) (string, error) {
+		value, executeErr := executor.Execute(t.Context(), "second", func(context.Context) (string, error) {
 			close(secondStarted)
 			return "two", nil
 		}, nil)
@@ -65,7 +65,7 @@ func TestExecutorPublishesResultBeforeCompletionCallbackInternal(t *testing.T) {
 	releaseAfter := make(chan struct{})
 	result := make(chan int, 1)
 	go func() {
-		value, _ := Execute(t.Context(), executor, "ordered", func(context.Context) (int, error) {
+		value, _ := executor.Execute(t.Context(), "ordered", func(context.Context) (int, error) {
 			return 42, nil
 		}, func(int, error) {
 			close(afterStarted)
@@ -91,12 +91,12 @@ func TestExecutorContainsTaskPanicAndContinuesInternal(t *testing.T) {
 	executor, err := NewExecutor(t.Context(), runtime, "executor-test", "panic", 1)
 	require.NoError(t, err)
 
-	_, err = Execute(t.Context(), executor, "panic task", func(context.Context) (NoPayload, error) {
+	_, err = executor.Execute(t.Context(), "panic task", func(context.Context) (NoPayload, error) {
 		panic("deliberate executor panic")
 	}, nil)
 	require.ErrorContains(t, err, "panic task panicked")
 
-	value, err := Execute(t.Context(), executor, "following task", func(context.Context) (int, error) {
+	value, err := executor.Execute(t.Context(), "following task", func(context.Context) (int, error) {
 		return 7, nil
 	}, nil)
 	require.NoError(t, err)

backend/internal/actors/gate.go

@@ -99,7 +99,7 @@ func (g *Gate[K]) TryAcquire(ctx context.Context, key K) (*Lease[K], bool, error
 
 	// Admission must resolve after it is queued; abandoning an admitted reply on
 	// caller cancellation would leak the key without a lease to release it.
-	admitted, err := awaitInternal(context.WithoutCancel(ctx), g.actor.handle, reply, errors.New("actor gate stopped"))
+	admitted, err := g.actor.handle.await(context.WithoutCancel(ctx), reply, errors.New("actor gate stopped"))
 	if err != nil {
 		return nil, false, err
 	}
@@ -125,7 +125,7 @@ func (l *Lease[K]) Release() {
 		}); err != nil {
 			return
 		}
-		_, _ = awaitInternal(context.Background(), l.gate.actor.handle, reply, errors.New("actor gate stopped"))
+		_, _ = l.gate.actor.handle.await(context.Background(), reply, errors.New("actor gate stopped"))
 	})
 }