feat(berths): bulk price update + per-berth price API

Two new endpoints lift price editing out of the full berth-update form:

- `PATCH /api/v1/berths/[id]/price` — single-berth price edit triggered
  inline from the berth list / detail (no need to open the heavy edit
  modal just to retag a price).
- `POST /api/v1/berths/bulk-update-prices` — multi-row update from a
  selection in the berth list; transactional, audit-logged per row.

Berth list column gets an inline price-edit affordance backed by the
single-berth endpoint; the bulk action lives in the row-selection
toolbar.

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

src/app/api/v1/berths/[id]/price/route.ts

@@ -0,0 +1,34 @@
+import { NextResponse } from 'next/server';
+
+import { withAuth, withPermission } from '@/lib/api/helpers';
+import { parseBody } from '@/lib/api/route-helpers';
+import { updateBerthPriceSchema } from '@/lib/validators/berths';
+import { updateBerthPrice } from '@/lib/services/berths.service';
+import { errorResponse } from '@/lib/errors';
+
+/**
+ * PATCH /api/v1/berths/[id]/price
+ *
+ * Focused price-update endpoint gated by the dedicated
+ * `berths.update_prices` permission. Lets a role mutate berth pricing
+ * without granting the full `berths.edit` surface.
+ *
+ * Always audited (one `audit_log` row per call with
+ * `field_changed='price'` and the before/after values).
+ */
+export const PATCH = withAuth(
+  withPermission('berths', 'update_prices', async (req, ctx, params) => {
+    try {
+      const body = await parseBody(req, updateBerthPriceSchema);
+      const updated = await updateBerthPrice(params.id!, ctx.portId, body, {
+        userId: ctx.userId,
+        portId: ctx.portId,
+        ipAddress: ctx.ipAddress,
+        userAgent: ctx.userAgent,
+      });
+      return NextResponse.json({ data: updated });
+    } catch (error) {
+      return errorResponse(error);
+    }
+  }),
+);

src/app/api/v1/berths/bulk-update-prices/route.ts

@@ -0,0 +1,35 @@
+import { NextResponse } from 'next/server';
+
+import { withAuth, withPermission } from '@/lib/api/helpers';
+import { parseBody } from '@/lib/api/route-helpers';
+import { bulkUpdateBerthPricesSchema } from '@/lib/validators/berths';
+import { bulkUpdateBerthPrices } from '@/lib/services/berths.service';
+import { errorResponse } from '@/lib/errors';
+
+/**
+ * POST /api/v1/berths/bulk-update-prices
+ *
+ * Bulk update berth prices in a single transaction (up to 500 per call).
+ * Gated by `berths.update_prices`. Returns counts so the UI can present
+ * "Updated N · Unchanged M · Missing K" feedback.
+ *
+ * Audit: one `audit_log` row per actually-updated berth (idempotent —
+ * berths whose new price matches the existing value are skipped and
+ * counted as `unchanged`).
+ */
+export const POST = withAuth(
+  withPermission('berths', 'update_prices', async (req, ctx) => {
+    try {
+      const body = await parseBody(req, bulkUpdateBerthPricesSchema);
+      const result = await bulkUpdateBerthPrices(ctx.portId, body, {
+        userId: ctx.userId,
+        portId: ctx.portId,
+        ipAddress: ctx.ipAddress,
+        userAgent: ctx.userAgent,
+      });
+      return NextResponse.json({ data: result });
+    } catch (error) {
+      return errorResponse(error);
+    }
+  }),
+);

src/components/berths/berth-columns.tsx

@@ -68,6 +68,9 @@ export type BerthRow = {
   /** Most-advanced pipeline stage among the berth's active interests. Null
    *  when no active interest is linked. Read-only; computed server-side. */
   latestInterestStage?: string | null;
+  /** Count of non-terminal, non-archived interests linked to this berth.
+   *  Drives the "Active interests" column + the demand sort. */
+  activeInterestCount?: number;
   /** #67: source of the last status write. 'manual' when a human set it
    *  via the API; 'automated' when a berth-rule fired; null on rows that
    *  haven't been touched since seed. The reconciliation surface treats
@@ -85,6 +88,7 @@ export const BERTH_COLUMN_OPTIONS: Array<{ id: string; label: string }> = [
   { id: 'area', label: 'Area' },
   { id: 'status', label: 'Status' },
   { id: 'latestInterestStage', label: 'Latest deal stage' },
+  { id: 'activeInterestCount', label: 'Active interests' },
   { id: 'sidePontoon', label: 'Side / Pontoon' },
   { id: 'dimensions', label: 'Dimensions' },
   { id: 'nominalBoatSize', label: 'Nominal boat size' },
@@ -281,6 +285,16 @@ export const berthColumns: ColumnDef<BerthRow, unknown>[] = [
       );
     },
   },
+  {
+    id: 'activeInterestCount',
+    accessorKey: 'activeInterestCount',
+    header: 'Active interests',
+    cell: ({ row }) => {
+      const n = row.original.activeInterestCount ?? 0;
+      if (n === 0) return <span className="text-muted-foreground">—</span>;
+      return <span className="font-medium tabular-nums">{n}</span>;
+    },
+  },
   {
     id: 'sidePontoon',
     header: 'Side / Pontoon',

src/lib/services/berths.service.ts

@@ -1,4 +1,4 @@
-import { and, desc, eq, gte, lte, inArray, isNull, notInArray, sql } from 'drizzle-orm';
+import { and, count, desc, eq, gte, lte, inArray, isNull, notInArray, sql } from 'drizzle-orm';
 
 import { db } from '@/lib/db';
 import { berths, berthTags, berthWaitingList, berthMaintenanceLog } from '@/lib/db/schema/berths';
@@ -19,6 +19,8 @@ import type {
   CreateBerthInput,
   UpdateBerthInput,
   UpdateBerthStatusInput,
+  UpdateBerthPriceInput,
+  BulkUpdateBerthPricesInput,
   ListBerthsQuery,
   AddMaintenanceLogInput,
   UpdateWaitingListInput,
@@ -89,6 +91,9 @@ export async function listBerths(portId: string, query: ListBerthsQuery) {
         return berths.status;
       case 'lengthM':
         return berths.lengthM;
+      case 'activeInterestCount':
+        // Sorted via correlated subquery in customOrderBy below.
+        return null;
       default:
         // No sort requested → natural mooring order is the friendliest
         // default for the berth grid (groups by pontoon letter).
@@ -96,6 +101,24 @@ export async function listBerths(portId: string, query: ListBerthsQuery) {
     }
   })();
 
+  // Sort by active interest count via correlated subquery. Cheap at
+  // marina scale (hundreds of berths × thousands of interests); revisit
+  // with a LATERAL join if multi-port reporting ever hits this hot.
+  const demandSort =
+    query.sort === 'activeInterestCount'
+      ? [
+          sql`(
+            SELECT COUNT(*)::int
+            FROM ${interestBerths} ib
+            INNER JOIN ${interests} i ON i.id = ib.interest_id
+            WHERE ib.berth_id = ${berths.id}
+              AND i.port_id = ${portId}
+              AND i.archived_at IS NULL
+              AND i.outcome IS NULL
+          ) ${sql.raw(query.order === 'asc' ? 'ASC' : 'DESC')}`,
+        ]
+      : null;
+
   const result = await buildListQuery({
     table: berths,
     portIdColumn: berths.portId,
@@ -104,7 +127,7 @@ export async function listBerths(portId: string, query: ListBerthsQuery) {
     updatedAtColumn: berths.updatedAt,
     filters,
     sort: sortColumn ? { column: sortColumn, direction: query.order } : undefined,
-    customOrderBy: sortColumn ? undefined : NATURAL_MOORING_SORT,
+    customOrderBy: demandSort ?? (sortColumn ? undefined : NATURAL_MOORING_SORT),
     page: query.page,
     pageSize: query.limit,
     searchColumns: [berths.mooringNumber, berths.area],
@@ -139,16 +162,44 @@ export async function listBerths(portId: string, query: ListBerthsQuery) {
   }
 
   const latestStageByBerthId = await getLatestInterestStageByBerth(berthIds, portId);
+  const interestCountByBerthId = await getActiveInterestCountByBerth(berthIds, portId);
 
   const data = (result.data as Array<Record<string, unknown>>).map((b) => ({
     ...b,
     tags: tagsByBerthId[b.id as string] ?? [],
     latestInterestStage: latestStageByBerthId[b.id as string] ?? null,
+    activeInterestCount: interestCountByBerthId[b.id as string] ?? 0,
   }));
 
   return { data, total: result.total };
 }
 
+/**
+ * Per-berth active interest count. Mirrors the demand-sort subquery so
+ * the column value and the sort key stay consistent.
+ */
+async function getActiveInterestCountByBerth(
+  berthIds: string[],
+  portId: string,
+): Promise<Record<string, number>> {
+  if (berthIds.length === 0) return {};
+  const rows = await db
+    .select({
+      berthId: interestBerths.berthId,
+      count: count(interests.id),
+    })
+    .from(interestBerths)
+    .innerJoin(interests, eq(interestBerths.interestId, interests.id))
+    .where(and(activeInterestsWhere(portId), inArray(interestBerths.berthId, berthIds)))
+    .groupBy(interestBerths.berthId);
+
+  const map: Record<string, number> = {};
+  for (const row of rows) {
+    map[row.berthId] = Number(row.count);
+  }
+  return map;
+}
+
 /**
  * For each berth id, returns the most-advanced pipeline stage among its
  * linked active interests (outcome IS NULL, not archived). Used by the
@@ -301,6 +352,167 @@ export async function updateBerth(
   return updated!;
 }
 
+// ─── Update Price (single + bulk) ─────────────────────────────────────────────
+
+/**
+ * Update a single berth's price (+ optional currency). Gated upstream by
+ * the `berths.update_prices` permission so sales reps can retune prices
+ * without the full `berths.edit` surface. Writes a focused audit row
+ * with `field_changed='price'` and the before/after values so the audit
+ * log carries a clean price-change history.
+ */
+export async function updateBerthPrice(
+  id: string,
+  portId: string,
+  data: UpdateBerthPriceInput,
+  meta: AuditMeta,
+) {
+  const existing = await db.query.berths.findFirst({
+    where: and(eq(berths.id, id), eq(berths.portId, portId)),
+  });
+  if (!existing) throw new NotFoundError('Berth');
+
+  const oldPrice = existing.price;
+  const oldCurrency = existing.priceCurrency;
+  const newPrice = data.price === null ? null : String(data.price);
+  const newCurrency = data.priceCurrency ?? oldCurrency;
+
+  if (oldPrice === newPrice && oldCurrency === newCurrency) {
+    return existing;
+  }
+
+  const [updated] = await db
+    .update(berths)
+    .set({
+      price: newPrice,
+      priceCurrency: newCurrency,
+      updatedAt: new Date(),
+    })
+    .where(and(eq(berths.id, id), eq(berths.portId, portId)))
+    .returning();
+
+  void createAuditLog({
+    userId: meta.userId,
+    portId,
+    action: 'update',
+    entityType: 'berth',
+    entityId: id,
+    fieldChanged: 'price',
+    oldValue: toAuditJson({ price: oldPrice, priceCurrency: oldCurrency }),
+    newValue: toAuditJson({ price: newPrice, priceCurrency: newCurrency }),
+    ipAddress: meta.ipAddress,
+    userAgent: meta.userAgent,
+  });
+
+  emitToRoom(`port:${portId}`, 'berth:updated', {
+    berthId: id,
+    changedFields: ['price'],
+  });
+
+  void import('@/lib/services/webhook-dispatch').then(({ dispatchWebhookEvent }) =>
+    dispatchWebhookEvent(portId, 'berth:updated', { berthId: id }),
+  );
+
+  return updated!;
+}
+
+/**
+ * Bulk berth price update. Loads all targeted rows in one query so we
+ * can validate port ownership in the same trip, then writes each change
+ * with a per-row audit entry. Berths that resolve to the same price are
+ * skipped (counted as `unchanged` in the response) so retries are idempotent.
+ *
+ * Returns counts so the UI can present "updated N, skipped M, missing K".
+ */
+export async function bulkUpdateBerthPrices(
+  portId: string,
+  data: BulkUpdateBerthPricesInput,
+  meta: AuditMeta,
+) {
+  const targetIds = data.updates.map((u) => u.berthId);
+  const existingRows = await db
+    .select({
+      id: berths.id,
+      price: berths.price,
+      priceCurrency: berths.priceCurrency,
+    })
+    .from(berths)
+    .where(and(eq(berths.portId, portId), inArray(berths.id, targetIds)));
+  const byId = new Map(existingRows.map((b) => [b.id, b]));
+
+  const missing: string[] = [];
+  const updatedIds: string[] = [];
+  const unchangedIds: string[] = [];
+
+  await db.transaction(async (tx) => {
+    for (const u of data.updates) {
+      const existing = byId.get(u.berthId);
+      if (!existing) {
+        missing.push(u.berthId);
+        continue;
+      }
+      const oldPrice = existing.price;
+      const oldCurrency = existing.priceCurrency;
+      const newPrice = u.price === null ? null : String(u.price);
+      const newCurrency = u.priceCurrency ?? oldCurrency;
+
+      if (oldPrice === newPrice && oldCurrency === newCurrency) {
+        unchangedIds.push(u.berthId);
+        continue;
+      }
+
+      await tx
+        .update(berths)
+        .set({
+          price: newPrice,
+          priceCurrency: newCurrency,
+          updatedAt: new Date(),
+        })
+        .where(and(eq(berths.id, u.berthId), eq(berths.portId, portId)));
+
+      void createAuditLog({
+        userId: meta.userId,
+        portId,
+        action: 'update',
+        entityType: 'berth',
+        entityId: u.berthId,
+        fieldChanged: 'price',
+        oldValue: toAuditJson({ price: oldPrice, priceCurrency: oldCurrency }),
+        newValue: toAuditJson({ price: newPrice, priceCurrency: newCurrency }),
+        metadata: { source: 'bulk_update_prices', batchSize: data.updates.length },
+        ipAddress: meta.ipAddress,
+        userAgent: meta.userAgent,
+      });
+
+      updatedIds.push(u.berthId);
+    }
+  });
+
+  // Realtime fan-out — one event per updated berth so any open list view
+  // refetches the affected rows.
+  for (const id of updatedIds) {
+    emitToRoom(`port:${portId}`, 'berth:updated', {
+      berthId: id,
+      changedFields: ['price'],
+    });
+  }
+
+  if (updatedIds.length > 0) {
+    void import('@/lib/services/webhook-dispatch').then(({ dispatchWebhookEvent }) => {
+      for (const id of updatedIds) {
+        dispatchWebhookEvent(portId, 'berth:updated', { berthId: id });
+      }
+    });
+  }
+
+  return {
+    updated: updatedIds.length,
+    unchanged: unchangedIds.length,
+    missing: missing.length,
+    missingIds: missing,
+  };
+}
+
 // ─── Update Status ────────────────────────────────────────────────────────────
 
 export async function updateBerthStatus(

src/lib/validators/berths.ts

@@ -120,6 +120,39 @@ export const listBerthsSchema = baseListQuerySchema.extend({
 
 export type ListBerthsQuery = z.infer<typeof listBerthsSchema>;
 
+// ─── Update Berth Price ───────────────────────────────────────────────────────
+// Single + bulk price update. Carved out from the general updateBerthSchema
+// so that callers with the `berths.update_prices` permission (but NOT the
+// generic `berths.edit`) can only mutate price + currency, not dimensions
+// or mooring metadata.
+
+export const updateBerthPriceSchema = z.object({
+  price: z.coerce.number().min(0, 'Price must be ≥ 0').nullable(),
+  priceCurrency: z
+    .string()
+    .trim()
+    .length(3, 'Currency must be a 3-letter ISO code')
+    .toUpperCase()
+    .optional(),
+});
+
+export type UpdateBerthPriceInput = z.infer<typeof updateBerthPriceSchema>;
+
+export const bulkUpdateBerthPricesSchema = z.object({
+  updates: z
+    .array(
+      z.object({
+        berthId: z.string().uuid(),
+        price: z.coerce.number().min(0).nullable(),
+        priceCurrency: z.string().trim().length(3).toUpperCase().optional(),
+      }),
+    )
+    .min(1, 'At least one berth required')
+    .max(500, 'Bulk price updates capped at 500 berths per call'),
+});
+
+export type BulkUpdateBerthPricesInput = z.infer<typeof bulkUpdateBerthPricesSchema>;
+
 // ─── Add Maintenance Log ──────────────────────────────────────────────────────
 
 export const addMaintenanceLogSchema = z.object({