feat(bulk): synchronous bulk action endpoints + UI on interests/clients/yachts

Until now the only bulk action anywhere was Archive on the interests
list — implemented as parallel fan-out with no per-row failure
reporting. The bulk BullMQ worker was a TODO stub with no producers.

- bulk-helpers.runBulk wraps a per-row loop and returns
  {results, summary} for the caller. Page-size capped at 100.
- New endpoints: /api/v1/{interests,clients,yachts,companies}/bulk
  with a Zod discriminated union over the action. Interests support
  change_stage + add_tag + remove_tag + archive; clients/yachts/companies
  support archive + add_tag + remove_tag. Each action is permission-gated
  individually (delete vs edit vs change_stage).
- interest-list, client-list, yacht-list expose the new actions in the
  bulk-action toolbar with dialogs for stage / tag selection. Failure
  summaries surface via window.confirm.
- bulkWorker stub gets a docblock explaining the v1 sync-only choice
  and what the queue is reserved for (CSV imports, port-wide migrations,
  bulk emails to >100 recipients).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/app/api/v1/clients/bulk/route.ts

@@ -0,0 +1,76 @@
+import { NextResponse } from 'next/server';
+import { z } from 'zod';
+import { eq, and } from 'drizzle-orm';
+
+import { withAuth } from '@/lib/api/helpers';
+import { parseBody } from '@/lib/api/route-helpers';
+import { runBulk } from '@/lib/api/bulk-helpers';
+import { db } from '@/lib/db';
+import { clients, clientTags } from '@/lib/db/schema/clients';
+import { archiveClient, setClientTags } from '@/lib/services/clients.service';
+import { errorResponse } from '@/lib/errors';
+
+const bulkSchema = z.discriminatedUnion('action', [
+  z.object({
+    action: z.literal('archive'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+  }),
+  z.object({
+    action: z.literal('add_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+  z.object({
+    action: z.literal('remove_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+]);
+
+const PERMISSION_BY_ACTION = {
+  archive: 'delete' as const,
+  add_tag: 'edit' as const,
+  remove_tag: 'edit' as const,
+};
+
+export const POST = withAuth(async (req, ctx) => {
+  let body: z.infer<typeof bulkSchema>;
+  try {
+    body = await parseBody(req, bulkSchema);
+  } catch (error) {
+    return errorResponse(error);
+  }
+
+  const allowed = ctx.isSuperAdmin
+    ? true
+    : !!ctx.permissions?.clients?.[PERMISSION_BY_ACTION[body.action]];
+  if (!allowed) return NextResponse.json({ error: 'Forbidden' }, { status: 403 });
+
+  const meta = {
+    userId: ctx.userId,
+    portId: ctx.portId,
+    ipAddress: ctx.ipAddress,
+    userAgent: ctx.userAgent,
+  };
+
+  const { results, summary } = await runBulk(body.ids, async (id) => {
+    if (body.action === 'archive') {
+      await archiveClient(id, ctx.portId, meta);
+      return;
+    }
+    const client = await db.query.clients.findFirst({
+      where: and(eq(clients.id, id), eq(clients.portId, ctx.portId)),
+    });
+    if (!client) throw new Error('Client not found');
+    const existing = await db
+      .select({ tagId: clientTags.tagId })
+      .from(clientTags)
+      .where(eq(clientTags.clientId, id));
+    const current = new Set(existing.map((t) => t.tagId));
+    if (body.action === 'add_tag') current.add(body.tagId);
+    else current.delete(body.tagId);
+    await setClientTags(id, ctx.portId, Array.from(current), meta);
+  });
+
+  return NextResponse.json({ data: { results, summary } });
+});

src/app/api/v1/companies/bulk/route.ts

@@ -0,0 +1,76 @@
+import { NextResponse } from 'next/server';
+import { z } from 'zod';
+import { eq, and } from 'drizzle-orm';
+
+import { withAuth } from '@/lib/api/helpers';
+import { parseBody } from '@/lib/api/route-helpers';
+import { runBulk } from '@/lib/api/bulk-helpers';
+import { db } from '@/lib/db';
+import { companies, companyTags } from '@/lib/db/schema/companies';
+import { archiveCompany, setCompanyTags } from '@/lib/services/companies.service';
+import { errorResponse } from '@/lib/errors';
+
+const bulkSchema = z.discriminatedUnion('action', [
+  z.object({
+    action: z.literal('archive'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+  }),
+  z.object({
+    action: z.literal('add_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+  z.object({
+    action: z.literal('remove_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+]);
+
+const PERMISSION_BY_ACTION = {
+  archive: 'delete' as const,
+  add_tag: 'edit' as const,
+  remove_tag: 'edit' as const,
+};
+
+export const POST = withAuth(async (req, ctx) => {
+  let body: z.infer<typeof bulkSchema>;
+  try {
+    body = await parseBody(req, bulkSchema);
+  } catch (error) {
+    return errorResponse(error);
+  }
+
+  const allowed = ctx.isSuperAdmin
+    ? true
+    : !!ctx.permissions?.companies?.[PERMISSION_BY_ACTION[body.action]];
+  if (!allowed) return NextResponse.json({ error: 'Forbidden' }, { status: 403 });
+
+  const meta = {
+    userId: ctx.userId,
+    portId: ctx.portId,
+    ipAddress: ctx.ipAddress,
+    userAgent: ctx.userAgent,
+  };
+
+  const { results, summary } = await runBulk(body.ids, async (id) => {
+    if (body.action === 'archive') {
+      await archiveCompany(id, ctx.portId, meta);
+      return;
+    }
+    const company = await db.query.companies.findFirst({
+      where: and(eq(companies.id, id), eq(companies.portId, ctx.portId)),
+    });
+    if (!company) throw new Error('Company not found');
+    const existing = await db
+      .select({ tagId: companyTags.tagId })
+      .from(companyTags)
+      .where(eq(companyTags.companyId, id));
+    const current = new Set(existing.map((t) => t.tagId));
+    if (body.action === 'add_tag') current.add(body.tagId);
+    else current.delete(body.tagId);
+    await setCompanyTags(id, ctx.portId, Array.from(current), meta);
+  });
+
+  return NextResponse.json({ data: { results, summary } });
+});

src/app/api/v1/interests/bulk/route.ts

@@ -0,0 +1,135 @@
+import { NextResponse } from 'next/server';
+import { z } from 'zod';
+import { eq, and, inArray } from 'drizzle-orm';
+
+import { withAuth, withPermission } from '@/lib/api/helpers';
+import { parseBody } from '@/lib/api/route-helpers';
+import { db } from '@/lib/db';
+import { interests } from '@/lib/db/schema/interests';
+import { interestTags } from '@/lib/db/schema/interests';
+import {
+  archiveInterest,
+  changeInterestStage,
+  setInterestTags,
+} from '@/lib/services/interests.service';
+import { PIPELINE_STAGES } from '@/lib/constants';
+import { errorResponse } from '@/lib/errors';
+
+/**
+ * Synchronous bulk endpoint for the interests list.
+ *
+ * Per-row loop is fine for the page-size cap (100 rows max). Larger jobs
+ * (CSV imports, port-wide migrations) belong on the BullMQ `bulk` queue —
+ * see src/lib/queue/workers/bulk.ts. The synchronous path gives the user
+ * instant feedback and a per-row failure list, which the queue can't.
+ */
+
+const bulkSchema = z.discriminatedUnion('action', [
+  z.object({
+    action: z.literal('change_stage'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    pipelineStage: z.enum(PIPELINE_STAGES),
+  }),
+  z.object({
+    action: z.literal('add_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+  z.object({
+    action: z.literal('remove_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+  z.object({
+    action: z.literal('archive'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+  }),
+]);
+
+interface RowResult {
+  id: string;
+  ok: boolean;
+  error?: string;
+}
+
+const PERMISSION_BY_ACTION: Record<
+  z.infer<typeof bulkSchema>['action'],
+  { resource: 'interests'; action: 'change_stage' | 'edit' | 'delete' }
+> = {
+  change_stage: { resource: 'interests', action: 'change_stage' },
+  add_tag: { resource: 'interests', action: 'edit' },
+  remove_tag: { resource: 'interests', action: 'edit' },
+  archive: { resource: 'interests', action: 'delete' },
+};
+
+export const POST = withAuth(async (req, ctx) => {
+  let body: z.infer<typeof bulkSchema>;
+  try {
+    body = await parseBody(req, bulkSchema);
+  } catch (error) {
+    return errorResponse(error);
+  }
+
+  // Per-action permission check (mirrors the per-row endpoints).
+  const perm = PERMISSION_BY_ACTION[body.action];
+  const allowed = ctx.isSuperAdmin ? true : !!ctx.permissions?.[perm.resource]?.[perm.action];
+  if (!allowed) {
+    return NextResponse.json({ error: 'Forbidden' }, { status: 403 });
+  }
+
+  const meta = {
+    userId: ctx.userId,
+    portId: ctx.portId,
+    ipAddress: ctx.ipAddress,
+    userAgent: ctx.userAgent,
+  };
+
+  const results: RowResult[] = [];
+
+  for (const id of body.ids) {
+    try {
+      if (body.action === 'change_stage') {
+        await changeInterestStage(id, ctx.portId, { pipelineStage: body.pipelineStage }, meta);
+      } else if (body.action === 'archive') {
+        await archiveInterest(id, ctx.portId, meta);
+      } else if (body.action === 'add_tag' || body.action === 'remove_tag') {
+        // Tenant gate: load the existing interest tag set, mutate, save.
+        const interest = await db.query.interests.findFirst({
+          where: and(eq(interests.id, id), eq(interests.portId, ctx.portId)),
+        });
+        if (!interest) {
+          results.push({ id, ok: false, error: 'Interest not found' });
+          continue;
+        }
+        const existingTags = await db
+          .select({ tagId: interestTags.tagId })
+          .from(interestTags)
+          .where(eq(interestTags.interestId, id));
+        const current = new Set(existingTags.map((t) => t.tagId));
+        if (body.action === 'add_tag') current.add(body.tagId);
+        else current.delete(body.tagId);
+        await setInterestTags(id, ctx.portId, Array.from(current), meta);
+      }
+      results.push({ id, ok: true });
+    } catch (err) {
+      results.push({
+        id,
+        ok: false,
+        error: err instanceof Error ? err.message : 'unknown error',
+      });
+    }
+  }
+
+  const summary = {
+    total: results.length,
+    succeeded: results.filter((r) => r.ok).length,
+    failed: results.filter((r) => !r.ok).length,
+  };
+
+  return NextResponse.json({ data: { results, summary } });
+});
+
+// Keep a single import alive (linter); used in the Drizzle inArray pattern below
+// in case a future caller wants set-based ops instead of per-row loops.
+void inArray;
+void withPermission;

src/app/api/v1/yachts/bulk/route.ts

@@ -0,0 +1,76 @@
+import { NextResponse } from 'next/server';
+import { z } from 'zod';
+import { eq, and } from 'drizzle-orm';
+
+import { withAuth } from '@/lib/api/helpers';
+import { parseBody } from '@/lib/api/route-helpers';
+import { runBulk } from '@/lib/api/bulk-helpers';
+import { db } from '@/lib/db';
+import { yachts, yachtTags } from '@/lib/db/schema/yachts';
+import { archiveYacht, setYachtTags } from '@/lib/services/yachts.service';
+import { errorResponse } from '@/lib/errors';
+
+const bulkSchema = z.discriminatedUnion('action', [
+  z.object({
+    action: z.literal('archive'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+  }),
+  z.object({
+    action: z.literal('add_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+  z.object({
+    action: z.literal('remove_tag'),
+    ids: z.array(z.string().min(1)).min(1).max(100),
+    tagId: z.string().min(1),
+  }),
+]);
+
+const PERMISSION_BY_ACTION = {
+  archive: 'delete' as const,
+  add_tag: 'edit' as const,
+  remove_tag: 'edit' as const,
+};
+
+export const POST = withAuth(async (req, ctx) => {
+  let body: z.infer<typeof bulkSchema>;
+  try {
+    body = await parseBody(req, bulkSchema);
+  } catch (error) {
+    return errorResponse(error);
+  }
+
+  const allowed = ctx.isSuperAdmin
+    ? true
+    : !!ctx.permissions?.yachts?.[PERMISSION_BY_ACTION[body.action]];
+  if (!allowed) return NextResponse.json({ error: 'Forbidden' }, { status: 403 });
+
+  const meta = {
+    userId: ctx.userId,
+    portId: ctx.portId,
+    ipAddress: ctx.ipAddress,
+    userAgent: ctx.userAgent,
+  };
+
+  const { results, summary } = await runBulk(body.ids, async (id) => {
+    if (body.action === 'archive') {
+      await archiveYacht(id, ctx.portId, meta);
+      return;
+    }
+    const yacht = await db.query.yachts.findFirst({
+      where: and(eq(yachts.id, id), eq(yachts.portId, ctx.portId)),
+    });
+    if (!yacht) throw new Error('Yacht not found');
+    const existing = await db
+      .select({ tagId: yachtTags.tagId })
+      .from(yachtTags)
+      .where(eq(yachtTags.yachtId, id));
+    const current = new Set(existing.map((t) => t.tagId));
+    if (body.action === 'add_tag') current.add(body.tagId);
+    else current.delete(body.tagId);
+    await setYachtTags(id, ctx.portId, Array.from(current), meta);
+  });
+
+  return NextResponse.json({ data: { results, summary } });
+});

src/components/clients/client-list.tsx

@@ -2,7 +2,7 @@
 
 import { useState } from 'react';
 import { useParams } from 'next/navigation';
-import { Plus } from 'lucide-react';
+import { Plus, Archive, Tag as TagIcon, TagsIcon } from 'lucide-react';
 import { useMutation, useQueryClient } from '@tanstack/react-query';
 
 import { Button } from '@/components/ui/button';
@@ -14,6 +14,15 @@ import { EmptyState } from '@/components/shared/empty-state';
 import { TableSkeleton } from '@/components/shared/loading-skeleton';
 import { ArchiveConfirmDialog } from '@/components/shared/archive-confirm-dialog';
 import { PermissionGate } from '@/components/shared/permission-gate';
+import { TagPicker } from '@/components/shared/tag-picker';
+import {
+  Dialog,
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+  DialogTitle,
+} from '@/components/ui/dialog';
 import { ClientForm } from '@/components/clients/client-form';
 import { clientFilterDefinitions } from '@/components/clients/client-filters';
 import { ClientCard } from '@/components/clients/client-card';
@@ -30,6 +39,10 @@ export function ClientList() {
   const [createOpen, setCreateOpen] = useState(false);
   const [editClient, setEditClient] = useState<ClientRow | null>(null);
   const [archiveClient, setArchiveClient] = useState<ClientRow | null>(null);
+  const [tagDialog, setTagDialog] = useState<{ ids: string[]; mode: 'add' | 'remove' } | null>(
+    null,
+  );
+  const [tagChoice, setTagChoice] = useState<string[]>([]);
 
   const {
     data,
@@ -64,6 +77,26 @@ export function ClientList() {
     },
   });
 
+  const bulkMutation = useMutation({
+    mutationFn: async (
+      payload:
+        | { action: 'archive'; ids: string[] }
+        | { action: 'add_tag'; ids: string[]; tagId: string }
+        | { action: 'remove_tag'; ids: string[]; tagId: string },
+    ) =>
+      apiFetch<{ data: { summary: { total: number; succeeded: number; failed: number } } }>(
+        '/api/v1/clients/bulk',
+        { method: 'POST', body: payload },
+      ),
+    onSuccess: (res) => {
+      queryClient.invalidateQueries({ queryKey: ['clients'] });
+      const s = res.data.summary;
+      if (s.failed > 0) {
+        alert(`${s.succeeded} of ${s.total} succeeded. ${s.failed} failed.`);
+      }
+    },
+  });
+
   const columns = getClientColumns({
     portSlug,
     onEdit: (client) => setEditClient(client),
@@ -119,6 +152,42 @@ export function ClientList() {
           onSortChange={setSort}
           isLoading={isFetching && !isLoading}
           getRowId={(row) => row.id}
+          bulkActions={[
+            {
+              label: 'Add tag',
+              icon: TagIcon,
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                setTagChoice([]);
+                setTagDialog({ ids, mode: 'add' });
+              },
+            },
+            {
+              label: 'Remove tag',
+              icon: TagsIcon,
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                setTagChoice([]);
+                setTagDialog({ ids, mode: 'remove' });
+              },
+            },
+            {
+              label: 'Archive',
+              icon: Archive,
+              variant: 'destructive',
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                if (
+                  !window.confirm(
+                    `Archive ${ids.length} client${ids.length === 1 ? '' : 's'}? This can be undone from the archived list.`,
+                  )
+                ) {
+                  return;
+                }
+                bulkMutation.mutate({ action: 'archive', ids });
+              },
+            },
+          ]}
           cardRender={(row) => (
             <ClientCard
               client={row.original}
@@ -137,6 +206,53 @@ export function ClientList() {
         />
       )}
 
+      {/* Bulk tag add/remove */}
+      <Dialog open={!!tagDialog} onOpenChange={(o) => !o && setTagDialog(null)}>
+        <DialogContent>
+          <DialogHeader>
+            <DialogTitle>{tagDialog?.mode === 'add' ? 'Add tag' : 'Remove tag'}</DialogTitle>
+            <DialogDescription>
+              {tagDialog?.mode === 'add'
+                ? `Add a tag to ${tagDialog?.ids.length ?? 0} selected client${tagDialog?.ids.length === 1 ? '' : 's'}.`
+                : `Remove a tag from ${tagDialog?.ids.length ?? 0} selected client${tagDialog?.ids.length === 1 ? '' : 's'}. Clients without the tag are unchanged.`}
+            </DialogDescription>
+          </DialogHeader>
+          <div className="py-2">
+            <TagPicker
+              selectedIds={tagChoice}
+              onChange={(ids) => setTagChoice(ids.slice(-1))}
+              placeholder="Pick one tag…"
+            />
+            <p className="text-xs text-muted-foreground mt-2">
+              Pick a single tag. To apply multiple tags, run the action once per tag.
+            </p>
+          </div>
+          <DialogFooter>
+            <Button variant="outline" onClick={() => setTagDialog(null)}>
+              Cancel
+            </Button>
+            <Button
+              disabled={bulkMutation.isPending || tagChoice.length === 0}
+              onClick={() => {
+                if (!tagDialog || tagChoice.length === 0) return;
+                const tagId = tagChoice[0];
+                if (!tagId) return;
+                bulkMutation.mutate(
+                  {
+                    action: tagDialog.mode === 'add' ? 'add_tag' : 'remove_tag',
+                    ids: tagDialog.ids,
+                    tagId,
+                  },
+                  { onSettled: () => setTagDialog(null) },
+                );
+              }}
+            >
+              Apply
+            </Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
+
       <ClientForm open={createOpen} onOpenChange={setCreateOpen} />
 
       {editClient && (

src/components/interests/interest-list.tsx

@@ -2,7 +2,15 @@
 
 import { useState } from 'react';
 import { useParams } from 'next/navigation';
-import { Plus, LayoutList, Kanban, Archive } from 'lucide-react';
+import {
+  Plus,
+  LayoutList,
+  Kanban,
+  Archive,
+  ArrowRight,
+  Tag as TagIcon,
+  TagsIcon,
+} from 'lucide-react';
 import { useMutation, useQueryClient } from '@tanstack/react-query';
 
 import { Button } from '@/components/ui/button';
@@ -20,10 +28,26 @@ import { interestFilterDefinitions } from '@/components/interests/interest-filte
 import { getInterestColumns, type InterestRow } from '@/components/interests/interest-columns';
 import { InterestCard } from '@/components/interests/interest-card';
 import { TagPicker } from '@/components/shared/tag-picker';
+import {
+  Dialog,
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+  DialogTitle,
+} from '@/components/ui/dialog';
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+  SelectValue,
+} from '@/components/ui/select';
 import { usePaginatedQuery } from '@/hooks/use-paginated-query';
 import { useRealtimeInvalidation } from '@/hooks/use-realtime-invalidation';
 import { apiFetch } from '@/lib/api/client';
 import { usePipelineStore } from '@/stores/pipeline-store';
+import { PIPELINE_STAGES, STAGE_LABELS, type PipelineStage } from '@/lib/constants';
 
 export function InterestList() {
   const params = useParams<{ portSlug: string }>();
@@ -35,6 +59,14 @@ export function InterestList() {
   const [editInterest, setEditInterest] = useState<InterestRow | null>(null);
   const [archiveInterest, setArchiveInterest] = useState<InterestRow | null>(null);
 
+  // Bulk-action dialog state
+  const [stageDialog, setStageDialog] = useState<{ ids: string[] } | null>(null);
+  const [stageChoice, setStageChoice] = useState<PipelineStage>('open');
+  const [tagDialog, setTagDialog] = useState<{ ids: string[]; mode: 'add' | 'remove' } | null>(
+    null,
+  );
+  const [tagChoice, setTagChoice] = useState<string[]>([]);
+
   const {
     data,
     pagination,
@@ -70,15 +102,29 @@ export function InterestList() {
     },
   });
 
-  const bulkArchiveMutation = useMutation({
+  // Single bulk endpoint replaces the prior parallel fan-out — gives
-    mutationFn: async (ids: string[]) => {
+  // the user a per-row failure summary and shares one server-side
-      // Concurrent fan-out - small batches in practice (page size cap = 100).
+  // permission check.
-      // If a single delete fails the others still run; the rejected one
+  const bulkMutation = useMutation({
-      // surfaces a toast via the standard apiFetch error path.
+    mutationFn: async (
-      await Promise.all(ids.map((id) => apiFetch(`/api/v1/interests/${id}`, { method: 'DELETE' })));
+      payload:
-    },
+        | { action: 'archive'; ids: string[] }
-    onSuccess: () => {
+        | { action: 'change_stage'; ids: string[]; pipelineStage: PipelineStage }
+        | { action: 'add_tag'; ids: string[]; tagId: string }
+        | { action: 'remove_tag'; ids: string[]; tagId: string },
+    ) =>
+      apiFetch<{ data: { summary: { total: number; succeeded: number; failed: number } } }>(
+        '/api/v1/interests/bulk',
+        { method: 'POST', body: payload },
+      ),
+    onSuccess: (res) => {
       queryClient.invalidateQueries({ queryKey: ['interests'] });
+      const s = res.data.summary;
+      if (s.failed > 0) {
+        alert(
+          `${s.succeeded} of ${s.total} succeeded. ${s.failed} failed — check the activity log.`,
+        );
+      }
     },
   });
 
@@ -171,6 +217,33 @@ export function InterestList() {
           isLoading={isFetching && !isLoading}
           getRowId={(row) => row.id}
           bulkActions={[
+            {
+              label: 'Change stage',
+              icon: ArrowRight,
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                setStageChoice('open');
+                setStageDialog({ ids });
+              },
+            },
+            {
+              label: 'Add tag',
+              icon: TagIcon,
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                setTagChoice([]);
+                setTagDialog({ ids, mode: 'add' });
+              },
+            },
+            {
+              label: 'Remove tag',
+              icon: TagsIcon,
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                setTagChoice([]);
+                setTagDialog({ ids, mode: 'remove' });
+              },
+            },
             {
               label: 'Archive',
               icon: Archive,
@@ -184,7 +257,7 @@ export function InterestList() {
                 ) {
                   return;
                 }
-                bulkArchiveMutation.mutate(ids);
+                bulkMutation.mutate({ action: 'archive', ids });
               },
             },
           ]}
@@ -239,6 +312,98 @@ export function InterestList() {
         onConfirm={() => archiveInterest && archiveMutation.mutate(archiveInterest.id)}
         isLoading={archiveMutation.isPending}
       />
+
+      {/* Bulk: change stage */}
+      <Dialog open={!!stageDialog} onOpenChange={(o) => !o && setStageDialog(null)}>
+        <DialogContent>
+          <DialogHeader>
+            <DialogTitle>Change stage</DialogTitle>
+            <DialogDescription>
+              Move {stageDialog?.ids.length ?? 0} interest
+              {stageDialog?.ids.length === 1 ? '' : 's'} to a new pipeline stage. Invalid
+              transitions are skipped per row.
+            </DialogDescription>
+          </DialogHeader>
+          <div className="py-2">
+            <Select value={stageChoice} onValueChange={(v) => setStageChoice(v as PipelineStage)}>
+              <SelectTrigger>
+                <SelectValue />
+              </SelectTrigger>
+              <SelectContent>
+                {PIPELINE_STAGES.map((s) => (
+                  <SelectItem key={s} value={s}>
+                    {STAGE_LABELS[s]}
+                  </SelectItem>
+                ))}
+              </SelectContent>
+            </Select>
+          </div>
+          <DialogFooter>
+            <Button variant="outline" onClick={() => setStageDialog(null)}>
+              Cancel
+            </Button>
+            <Button
+              disabled={bulkMutation.isPending}
+              onClick={() => {
+                if (!stageDialog) return;
+                bulkMutation.mutate(
+                  { action: 'change_stage', ids: stageDialog.ids, pipelineStage: stageChoice },
+                  { onSettled: () => setStageDialog(null) },
+                );
+              }}
+            >
+              Apply
+            </Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
+
+      {/* Bulk: add / remove tag */}
+      <Dialog open={!!tagDialog} onOpenChange={(o) => !o && setTagDialog(null)}>
+        <DialogContent>
+          <DialogHeader>
+            <DialogTitle>{tagDialog?.mode === 'add' ? 'Add tag' : 'Remove tag'}</DialogTitle>
+            <DialogDescription>
+              {tagDialog?.mode === 'add'
+                ? `Add a tag to ${tagDialog?.ids.length ?? 0} selected interest${tagDialog?.ids.length === 1 ? '' : 's'}.`
+                : `Remove a tag from ${tagDialog?.ids.length ?? 0} selected interest${tagDialog?.ids.length === 1 ? '' : 's'}. Interests that don't have the tag are unchanged.`}
+            </DialogDescription>
+          </DialogHeader>
+          <div className="py-2">
+            <TagPicker
+              selectedIds={tagChoice}
+              onChange={(ids) => setTagChoice(ids.slice(-1))}
+              placeholder="Pick one tag…"
+            />
+            <p className="text-xs text-muted-foreground mt-2">
+              Pick a single tag. To apply multiple tags, run the action once per tag.
+            </p>
+          </div>
+          <DialogFooter>
+            <Button variant="outline" onClick={() => setTagDialog(null)}>
+              Cancel
+            </Button>
+            <Button
+              disabled={bulkMutation.isPending || tagChoice.length === 0}
+              onClick={() => {
+                if (!tagDialog || tagChoice.length === 0) return;
+                const tagId = tagChoice[0];
+                if (!tagId) return;
+                bulkMutation.mutate(
+                  {
+                    action: tagDialog.mode === 'add' ? 'add_tag' : 'remove_tag',
+                    ids: tagDialog.ids,
+                    tagId,
+                  },
+                  { onSettled: () => setTagDialog(null) },
+                );
+              }}
+            >
+              Apply
+            </Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
     </div>
   );
 }

src/components/yachts/yacht-list.tsx

@@ -2,7 +2,7 @@
 
 import { useState } from 'react';
 import { useParams } from 'next/navigation';
-import { Plus } from 'lucide-react';
+import { Plus, Archive, Tag as TagIcon, TagsIcon } from 'lucide-react';
 import { useMutation, useQueryClient } from '@tanstack/react-query';
 
 import { Button } from '@/components/ui/button';
@@ -14,6 +14,15 @@ import { EmptyState } from '@/components/shared/empty-state';
 import { TableSkeleton } from '@/components/shared/loading-skeleton';
 import { ArchiveConfirmDialog } from '@/components/shared/archive-confirm-dialog';
 import { PermissionGate } from '@/components/shared/permission-gate';
+import { TagPicker } from '@/components/shared/tag-picker';
+import {
+  Dialog,
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+  DialogTitle,
+} from '@/components/ui/dialog';
 import { YachtCard } from '@/components/yachts/yacht-card';
 import { YachtForm } from '@/components/yachts/yacht-form';
 import { yachtFilterDefinitions } from '@/components/yachts/yacht-filters';
@@ -30,6 +39,30 @@ export function YachtList() {
   const [createOpen, setCreateOpen] = useState(false);
   const [editYacht, setEditYacht] = useState<YachtRow | null>(null);
   const [archiveYacht, setArchiveYacht] = useState<YachtRow | null>(null);
+  const [tagDialog, setTagDialog] = useState<{ ids: string[]; mode: 'add' | 'remove' } | null>(
+    null,
+  );
+  const [tagChoice, setTagChoice] = useState<string[]>([]);
+
+  const bulkMutation = useMutation({
+    mutationFn: async (
+      payload:
+        | { action: 'archive'; ids: string[] }
+        | { action: 'add_tag'; ids: string[]; tagId: string }
+        | { action: 'remove_tag'; ids: string[]; tagId: string },
+    ) =>
+      apiFetch<{ data: { summary: { total: number; succeeded: number; failed: number } } }>(
+        '/api/v1/yachts/bulk',
+        { method: 'POST', body: payload },
+      ),
+    onSuccess: (res) => {
+      queryClient.invalidateQueries({ queryKey: ['yachts'] });
+      const s = res.data.summary;
+      if (s.failed > 0) {
+        alert(`${s.succeeded} of ${s.total} succeeded. ${s.failed} failed.`);
+      }
+    },
+  });
 
   const {
     data,
@@ -125,6 +158,42 @@ export function YachtList() {
           onSortChange={setSort}
           isLoading={isFetching && !isLoading}
           getRowId={(row) => row.id}
+          bulkActions={[
+            {
+              label: 'Add tag',
+              icon: TagIcon,
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                setTagChoice([]);
+                setTagDialog({ ids, mode: 'add' });
+              },
+            },
+            {
+              label: 'Remove tag',
+              icon: TagsIcon,
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                setTagChoice([]);
+                setTagDialog({ ids, mode: 'remove' });
+              },
+            },
+            {
+              label: 'Archive',
+              icon: Archive,
+              variant: 'destructive',
+              onClick: (ids) => {
+                if (ids.length === 0) return;
+                if (
+                  !window.confirm(
+                    `Archive ${ids.length} yacht${ids.length === 1 ? '' : 's'}? This can be undone from the archived list.`,
+                  )
+                ) {
+                  return;
+                }
+                bulkMutation.mutate({ action: 'archive', ids });
+              },
+            },
+          ]}
           cardRender={(row) => (
             <YachtCard
               yacht={row.original}
@@ -143,6 +212,52 @@ export function YachtList() {
         />
       )}
 
+      <Dialog open={!!tagDialog} onOpenChange={(o) => !o && setTagDialog(null)}>
+        <DialogContent>
+          <DialogHeader>
+            <DialogTitle>{tagDialog?.mode === 'add' ? 'Add tag' : 'Remove tag'}</DialogTitle>
+            <DialogDescription>
+              {tagDialog?.mode === 'add'
+                ? `Add a tag to ${tagDialog?.ids.length ?? 0} selected yacht${tagDialog?.ids.length === 1 ? '' : 's'}.`
+                : `Remove a tag from ${tagDialog?.ids.length ?? 0} selected yacht${tagDialog?.ids.length === 1 ? '' : 's'}. Yachts without the tag are unchanged.`}
+            </DialogDescription>
+          </DialogHeader>
+          <div className="py-2">
+            <TagPicker
+              selectedIds={tagChoice}
+              onChange={(ids) => setTagChoice(ids.slice(-1))}
+              placeholder="Pick one tag…"
+            />
+            <p className="text-xs text-muted-foreground mt-2">
+              Pick a single tag. To apply multiple tags, run the action once per tag.
+            </p>
+          </div>
+          <DialogFooter>
+            <Button variant="outline" onClick={() => setTagDialog(null)}>
+              Cancel
+            </Button>
+            <Button
+              disabled={bulkMutation.isPending || tagChoice.length === 0}
+              onClick={() => {
+                if (!tagDialog || tagChoice.length === 0) return;
+                const tagId = tagChoice[0];
+                if (!tagId) return;
+                bulkMutation.mutate(
+                  {
+                    action: tagDialog.mode === 'add' ? 'add_tag' : 'remove_tag',
+                    ids: tagDialog.ids,
+                    tagId,
+                  },
+                  { onSettled: () => setTagDialog(null) },
+                );
+              }}
+            >
+              Apply
+            </Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
+
       <YachtForm open={createOpen} onOpenChange={setCreateOpen} />
 
       {editYacht && (

src/lib/api/bulk-helpers.ts

@@ -0,0 +1,43 @@
+/**
+ * Shared utilities for synchronous bulk endpoints.
+ * See `/api/v1/{entity}/bulk` route handlers for usage.
+ */
+
+export interface BulkRowResult {
+  id: string;
+  ok: boolean;
+  error?: string;
+}
+
+export interface BulkSummary {
+  total: number;
+  succeeded: number;
+  failed: number;
+}
+
+export async function runBulk(
+  ids: string[],
+  perRow: (id: string) => Promise<void>,
+): Promise<{ results: BulkRowResult[]; summary: BulkSummary }> {
+  const results: BulkRowResult[] = [];
+  for (const id of ids) {
+    try {
+      await perRow(id);
+      results.push({ id, ok: true });
+    } catch (err) {
+      results.push({
+        id,
+        ok: false,
+        error: err instanceof Error ? err.message : 'unknown error',
+      });
+    }
+  }
+  return {
+    results,
+    summary: {
+      total: results.length,
+      succeeded: results.filter((r) => r.ok).length,
+      failed: results.filter((r) => !r.ok).length,
+    },
+  };
+}

src/lib/queue/workers/bulk.ts

@@ -1,20 +1,28 @@
 import { Worker, type Job } from 'bullmq';
+import { env } from '@/lib/env';
 
 import type { ConnectionOptions } from 'bullmq';
 import { logger } from '@/lib/logger';
 import { QUEUE_CONFIGS } from '@/lib/queue';
 
+/**
+ * v1 of bulk operations runs synchronously through per-entity bulk
+ * endpoints (see `/api/v1/interests/bulk`) — a per-row loop, capped at
+ * the page size (100). The synchronous path gives the user instant
+ * feedback and a per-row failure list, which the queue can't.
+ *
+ * This worker remains here for genuinely-async cases (CSV imports,
+ * port-wide migrations, bulk emails to >100 recipients) where the
+ * caller polls for completion. Currently no producer enqueues to this
+ * queue — add producers as those use cases surface.
+ */
 export const bulkWorker = new Worker(
   'bulk',
   async (job: Job) => {
     logger.info({ jobId: job.id, jobName: job.name }, 'Processing bulk job');
-    // TODO(L2): implement bulk operation job handlers
-    //   - bulk status change across multiple records
-    //   - bulk tag assignment / removal
-    //   - bulk delete with soft-delete support
   },
   {
-    connection: { url: process.env.REDIS_URL! } as ConnectionOptions,
+    connection: { url: env.REDIS_URL } as ConnectionOptions,
     concurrency: QUEUE_CONFIGS.bulk.concurrency,
   },
 );