feat(dashboard): worker label filters, webhook 429 on limit exhausted

api-contracts/openapi/paths/v1/webhooks/webhook.yaml

@@ -140,6 +140,12 @@ V1WebhookGetDeleteReceiveUpdate:
             schema:
               $ref: "../../../components/schemas/_index.yaml#/APIErrors"
         description: Forbidden
+      "429":
+        content:
+          application/json:
+            schema:
+              $ref: "../../../components/schemas/_index.yaml#/APIErrors"
+        description: Resource limit exceeded for the tenant
     summary: Post a webhook message
     tags:
       - Webhook

api-contracts/openapi/paths/worker/worker.yaml

@@ -35,6 +35,16 @@ withTenant:
           type: array
           items:
             $ref: "../../components/schemas/_index.yaml#/WorkerStatus"
+      - description: Filter by worker labels
+        in: query
+        name: labels
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
+            description: The label key-value pairs (delimited by a `:`) to filter by
+            minLength: 1
     responses:
       "200":
         content:

api/v1/server/handlers/v1/webhooks/receive.go

@@ -321,6 +321,16 @@ func (w *V1WebhooksService) V1WebhookReceive(ctx echo.Context, request gen.V1Web
 	)
 
 	if err != nil {
+		if errors.Is(err, repository.ErrResourceExhausted) {
+			return gen.V1WebhookReceive429JSONResponse{
+				Errors: []gen.APIError{
+					{
+						Description: "resource limit exceeded: task run or event limit reached for tenant",
+					},
+				},
+			}, nil
+		}
+
 		return nil, fmt.Errorf("failed to ingest event")
 	}
 

api/v1/server/handlers/workers/list.go

@@ -3,11 +3,13 @@ package workers
 import (
 	"fmt"
 	"math"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
 
+	"github.com/hatchet-dev/hatchet/api/v1/server/oas/apierrors"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers"
 	transformersv1 "github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers/v1"
@@ -30,6 +32,25 @@ func (t *WorkerService) WorkerList(ctx echo.Context, request gen.WorkerListReque
 	}
 }
 
+func parseLabelFilters(labels *[]string) (keys []string, values []string, err error) {
+	if labels == nil {
+		return nil, nil, nil
+	}
+
+	for _, l := range *labels {
+		split := strings.SplitN(l, ":", 2)
+
+		if len(split) != 2 || split[0] == "" || split[1] == "" {
+			return nil, nil, fmt.Errorf("invalid label filter format: %s, expected key:value", l)
+		}
+
+		keys = append(keys, split[0])
+		values = append(values, split[1])
+	}
+
+	return keys, values, nil
+}
+
 func (t *WorkerService) workerListV0(ctx echo.Context, tenant *sqlcv1.Tenant, request gen.WorkerListRequestObject) (gen.WorkerListResponseObject, error) {
 	reqCtx := ctx.Request().Context()
 	tenantId := tenant.ID
@@ -63,6 +84,13 @@ func (t *WorkerService) workerListV0(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		opts.Statuses = statuses
 	}
 
+	labelKeys, labelValues, err := parseLabelFilters(request.Params.Labels)
+	if err != nil {
+		return gen.WorkerList400JSONResponse(apierrors.NewAPIErrors(err.Error())), nil
+	}
+	opts.LabelKeys = labelKeys
+	opts.LabelValues = labelValues
+
 	_, listSpan := telemetry.NewSpan(reqCtx, "worker-service.v0.list-workers")
 	defer listSpan.End()
 
@@ -152,6 +180,13 @@ func (t *WorkerService) workerListV1(ctx echo.Context, tenant *sqlcv1.Tenant, re
 		opts.Statuses = statuses
 	}
 
+	labelKeys, labelValues, err := parseLabelFilters(request.Params.Labels)
+	if err != nil {
+		return gen.WorkerList400JSONResponse(apierrors.NewAPIErrors(err.Error())), nil
+	}
+	opts.LabelKeys = labelKeys
+	opts.LabelValues = labelValues
+
 	listCtx, listSpan := telemetry.NewSpan(reqCtx, "worker-service.v1.list-workers")
 	defer listSpan.End()
 

frontend/app/src/components/v1/molecules/data-table/data-table-toolbar.tsx

@@ -119,7 +119,7 @@ export function DataTableToolbar<TData>({
 
   return (
     <div className="flex items-center justify-between">
-      <div className="flex w-full flex-row items-center gap-2">
+      <div className="flex w-full flex-row items-center gap-2 overflow-x-auto">
         <div className="flex min-w-0 flex-shrink-0 items-center gap-2">
           {!leftActions && isLoading && <Spinner />}
           {leftActions}

frontend/app/src/components/v1/ui/command.tsx

@@ -60,13 +60,23 @@ CommandInput.displayName = CommandPrimitive.Input.displayName;
 const CommandList = React.forwardRef<
   React.ElementRef<typeof CommandPrimitive.List>,
   React.ComponentPropsWithoutRef<typeof CommandPrimitive.List>
->(({ className, ...props }, ref) => (
+>(({ className, onWheel, ...props }, ref) => (
   <CommandPrimitive.List
     ref={ref}
     className={cn(
       'max-h-[300px] overflow-y-auto overflow-x-hidden overscroll-contain touch-pan-y [-webkit-overflow-scrolling:touch]',
       className,
     )}
+    // Popovers (e.g. the trigger workflow modal -> select workflow dropdown) render their content in a portal outside
+    // of any enclosing modal Dialog's content. Radix Dialog's scroll lock
+    // (react-remove-scroll) installs a document-level wheel listener that calls
+    // preventDefault() on wheel events outside the dialog's content/shards, which
+    // blocks trackpad/wheel scrolling here (scrollbar drag still works since that
+    // doesn't dispatch wheel events). Stop the event from bubbling to that listener.
+    onWheel={(e) => {
+      e.stopPropagation();
+      onWheel?.(e);
+    }}
     {...props}
   />
 ));

frontend/app/src/pages/error/components/new-version-available.tsx

@@ -9,7 +9,9 @@ export function NewVersionAvailable() {
   const navigate = useNavigate();
   const location = useLocation();
 
-  const queryParams = new URLSearchParams(location.search);
+  const queryParams = new URLSearchParams(
+    location.search as Record<string, string>,
+  );
 
   if (!queryParams.has('updated')) {
     queryParams.set('updated', 'true');

frontend/app/src/pages/main/v1/workers/components/worker-columns.tsx

@@ -28,7 +28,7 @@ const startedAtKey: WorkerColumnKeys = 'startedAt';
 const slotsKey: WorkerColumnKeys = 'slots';
 const lastHeartbeatAtKey: WorkerColumnKeys = 'lastHeartbeatAt';
 const runtimeKey: WorkerColumnKeys = 'runtime';
-const labelsKey: WorkerColumnKeys = 'labels';
+export const labelsKey: WorkerColumnKeys = 'labels';
 
 interface WorkerStatusBadgeProps extends BadgeProps {
   status?: string;

frontend/app/src/pages/main/v1/workers/index.tsx

@@ -1,4 +1,9 @@
-import { columns, statusKey, WorkerColumn } from './components/worker-columns';
+import {
+  columns,
+  labelsKey,
+  statusKey,
+  WorkerColumn,
+} from './components/worker-columns';
 import { ToolbarType } from '@/components/v1/molecules/data-table/data-table-toolbar';
 import { DataTable } from '@/components/v1/molecules/data-table/data-table.tsx';
 import { EmptyState } from '@/components/v1/molecules/empty-state/empty-state';
@@ -21,10 +26,12 @@ import { z } from 'zod';
 const workersQuerySchema = z
   .object({
     s: z.array(z.enum(['ACTIVE', 'INACTIVE', 'PAUSED'])).optional(), // status
+    l: z.array(z.string()).optional(), // labels
   })
   .default({})
   .transform((data) => ({
     s: data.s ?? ['ACTIVE', 'PAUSED'],
+    l: data.l ?? [],
   }));
 
 export default function Workers() {
@@ -52,16 +59,19 @@ function WorkersTable() {
   );
 
   const {
-    state: { s: statuses },
+    state: { s: statuses, l: labels },
     columnFilters,
     setColumnFilters,
     resetFilters,
-  } = useZodColumnFilters(workersQuerySchema, paramKey, { s: statusKey });
+  } = useZodColumnFilters(workersQuerySchema, paramKey, {
+    s: statusKey,
+    l: labelsKey,
+  });
 
   const { pagination, setPagination, limit, offset, setPageSize } =
     usePagination({
       key: paramKey,
-      resetPageOnChange: [statuses],
+      resetPageOnChange: [statuses, labels],
     });
 
   const [columnVisibility, setColumnVisibility] =
@@ -82,6 +92,7 @@ function WorkersTable() {
       offset,
       limit,
       statuses: statuses as WorkerStatus[],
+      labels: labels.length > 0 ? labels : undefined,
     }),
     refetchInterval,
   });
@@ -110,6 +121,11 @@ function WorkersTable() {
             { value: 'INACTIVE', label: 'Inactive' },
           ],
         },
+        {
+          columnId: labelsKey,
+          title: 'Labels',
+          type: ToolbarType.KeyValue,
+        },
       ]}
       emptyState={
         <EmptyState

frontend/app/src/pages/main/v1/workflow-runs-v1/$run/index.tsx

@@ -194,7 +194,7 @@ export default function Run() {
 
   if (
     taskRunQuery.isError &&
-    wasRedirectedFromTrigger === 'true' &&
+    wasRedirectedFromTrigger === true &&
     getErrorStatus(taskRunQuery.error) === 404
   ) {
     seenRedirect404.current = true;

frontend/app/src/pages/main/v1/workflows/$workflow/components/trigger-workflow-form.tsx

@@ -179,7 +179,7 @@ export function TriggerWorkflowForm({
       navigate({
         to: appRoutes.tenantRunRoute.to,
         params: { tenant: tenantId, run: workflowRun.run.metadata.id },
-        search: (prev) => ({ ...prev, wasRedirectedFromTrigger: 'true' }),
+        search: (prev) => ({ ...prev, wasRedirectedFromTrigger: true }),
       });
     },
     onError: handleApiError,

frontend/app/src/router.tsx

@@ -613,7 +613,7 @@ const tenantRunsRoute = createRoute({
 });
 
 const runSearchSchema = z.object({
-  wasRedirectedFromTrigger: z.enum(['true', 'false']).optional(), // hack to preserve typing elsewhere
+  wasRedirectedFromTrigger: z.boolean().optional(),
 });
 
 export const tenantRunRoute = createRoute({

pkg/repository/sqlcv1/workers.sql

@@ -150,6 +150,29 @@ WHERE
             ELSE 'ACTIVE'
         END = ANY(sqlc.narg('statuses')::text[])
     )
+    AND (
+        sqlc.narg('labelKeys')::text[] IS NULL
+        OR sqlc.narg('labelValues')::text[] IS NULL
+        OR (
+            SELECT BOOL_AND(
+                EXISTS (
+                    SELECT 1
+                    FROM "WorkerLabel" wl
+                    WHERE wl."workerId" = workers."id"
+                        AND wl."key" = lf.k
+                        AND (
+                            wl."strValue" = lf.v
+                            OR wl."intValue"::text = lf.v
+                        )
+                )
+            )
+            FROM (
+                SELECT
+                    UNNEST(sqlc.narg('labelKeys')::text[]) AS k,
+                    UNNEST(sqlc.narg('labelValues')::text[]) AS v
+            ) AS lf
+        )
+    )
 ORDER BY
     workers."createdAt" DESC
 OFFSET
@@ -195,6 +218,29 @@ WHERE
             WHEN workers."isPaused" = true THEN 'PAUSED'
             ELSE 'ACTIVE'
         END = ANY(sqlc.narg('statuses')::text[])
+    )
+    AND (
+        sqlc.narg('labelKeys')::text[] IS NULL
+        OR sqlc.narg('labelValues')::text[] IS NULL
+        OR (
+            SELECT BOOL_AND(
+                EXISTS (
+                    SELECT 1
+                    FROM "WorkerLabel" wl
+                    WHERE wl."workerId" = workers."id"
+                        AND wl."key" = lf.k
+                        AND (
+                            wl."strValue" = lf.v
+                            OR wl."intValue"::text = lf.v
+                        )
+                )
+            )
+            FROM (
+                SELECT
+                    UNNEST(sqlc.narg('labelKeys')::text[]) AS k,
+                    UNNEST(sqlc.narg('labelValues')::text[]) AS v
+            ) AS lf
+        )
     );
 
 -- name: GetWorkerById :one