Merge remote-tracking branch 'origin/main' into fix/issue-211

admin/src/app/dashboard/CampaignsPage.tsx

@@ -407,8 +407,9 @@ function StopCampaignDialog({
                     <DialogTitle>Force-stop campaign</DialogTitle>
                     <DialogDescription>
                         Stopping <span className="font-mono">{campaign.name}</span>.
-                        This is logged to the audit trail and the campaign owner
-                        will see the campaign status change to stopped.
+                        The campaign is paused, so the owner can see why and
+                        restart it once it is fixed. The reason is written to the
+                        audit trail and to the campaign's own activity feed.
                     </DialogDescription>
                 </DialogHeader>
                 <div>

admin/src/app/dashboard/UserRateLimitsDialog.tsx

@@ -1,7 +1,5 @@
-// User rate-limit override editor. Same "0 = inherit, positive = explicit
-// override" convention as the org limit overrides — leaving a field
-// blank means "don't touch it" on the PATCH, and the GET endpoint
-// already returns null for any unset value.
+// User rate-limit override editor. user_rate_limits has no null state, so a
+// blank field means "leave it alone" and clearing means typing the default back.
 
 import { useEffect, useState } from "react";
 import { useMutation, useQueryClient } from "@tanstack/react-query";
@@ -24,14 +22,26 @@ import type {
 } from "@/lib/api/models/admin";
 
 type FieldKey =
-    | "daily_email_limit"
+    | "limit_read_pm"
+    | "limit_write_pm"
+    | "limit_bulk_pm"
+    | "limit_unibox_pm"
+    | "limit_analytics_pm"
+    | "limit_api_calls_daily"
+    | "limit_bulk_ops_daily"
     | "max_connections"
     | "limit_ws_message_pm"
     | "limit_ws_join_pm"
     | "limit_ws_event_pm";
 
 const FIELDS: { key: FieldKey; label: string; hint: string }[] = [
-    { key: "daily_email_limit", label: "Daily emails", hint: "Outbound mail per day" },
+    { key: "limit_read_pm", label: "Reads / min", hint: "GET-style API calls per minute" },
+    { key: "limit_write_pm", label: "Writes / min", hint: "Mutating API calls per minute" },
+    { key: "limit_bulk_pm", label: "Bulk / min", hint: "Bulk import and export calls per minute" },
+    { key: "limit_unibox_pm", label: "Unibox / min", hint: "Unified inbox calls per minute" },
+    { key: "limit_analytics_pm", label: "Analytics / min", hint: "Analytics calls per minute" },
+    { key: "limit_api_calls_daily", label: "API calls / day", hint: "Total API calls per day" },
+    { key: "limit_bulk_ops_daily", label: "Bulk ops / day", hint: "Bulk operations per day" },
     { key: "max_connections", label: "Max connections", hint: "Concurrent websocket sessions" },
     { key: "limit_ws_message_pm", label: "WS messages / min", hint: "Realtime messages per minute" },
     { key: "limit_ws_join_pm", label: "WS joins / min", hint: "Realtime channel joins per minute" },
@@ -52,11 +62,12 @@ export function UserRateLimitsDialog({
     onOpenChange: (v: boolean) => void;
 }) {
     const qc = useQueryClient();
-    const [form, setForm] = useState<Record<FieldKey, string>>(() => seedForm(current));
+    const [form, setForm] = useState<Record<FieldKey, string>>(blankForm);
 
+    // Reset on open only: refetching `current` mid-edit must not wipe the form.
     useEffect(() => {
-        if (open) setForm(seedForm(current));
-    }, [open, current]);
+        if (open) setForm(blankForm());
+    }, [open]);
 
     const mutation = useMutation({
         mutationFn: (req: UpdateUserRateLimitsRequest) => updateUserRateLimits(userId, req),
@@ -97,9 +108,10 @@ export function UserRateLimitsDialog({
                     <DialogTitle>Rate limit overrides</DialogTitle>
                     <DialogDescription>
                         Per-user overrides for{" "}
-                        <span className="font-mono">{userEmail}</span>. Leave blank
-                        to keep the current value, set to <strong>0</strong> to
-                        clear the override (back to plan default).
+                        <span className="font-mono">{userEmail}</span>. Leave a
+                        field blank to keep its current value. There is no null
+                        state here, so <strong>0</strong> means zero, not
+                        "inherit".
                     </DialogDescription>
                 </DialogHeader>
 
@@ -154,20 +166,10 @@ export function UserRateLimitsDialog({
     );
 }
 
-function seedForm(current: AdminUserRateLimits | null | undefined): Record<FieldKey, string> {
-    const blank: Record<FieldKey, string> = {
-        daily_email_limit: "",
-        max_connections: "",
-        limit_ws_message_pm: "",
-        limit_ws_join_pm: "",
-        limit_ws_event_pm: "",
-    };
-    if (!current) return blank;
-    return {
-        daily_email_limit: current.daily_email_limit != null ? String(current.daily_email_limit) : "",
-        max_connections: current.max_connections != null ? String(current.max_connections) : "",
-        limit_ws_message_pm: current.limit_ws_message_pm != null ? String(current.limit_ws_message_pm) : "",
-        limit_ws_join_pm: current.limit_ws_join_pm != null ? String(current.limit_ws_join_pm) : "",
-        limit_ws_event_pm: current.limit_ws_event_pm != null ? String(current.limit_ws_event_pm) : "",
-    };
+// Blank on purpose: the current value is shown beside the input, not in it.
+function blankForm(): Record<FieldKey, string> {
+    return FIELDS.reduce(
+        (acc, f) => ({ ...acc, [f.key]: "" }),
+        {} as Record<FieldKey, string>,
+    );
 }

admin/src/lib/api/client/admin/users.ts

@@ -71,7 +71,8 @@ export function unbanUser(id: string, body: UnbanUserRequest): Promise<void> {
     });
 }
 
-export function getUserRateLimits(id: string): Promise<AdminUserRateLimits | null> {
+// Always resolves: a user with no override row reads back the product defaults.
+export function getUserRateLimits(id: string): Promise<AdminUserRateLimits> {
     return Request({
         method: "GET",
         url: `/admin/users/${id}/rate-limits`,

admin/src/lib/api/models/admin.ts

@@ -62,7 +62,7 @@ export interface AdminWorkerEmail {
     status: string;
     provider: string;
     warmup_enabled: boolean;
-    last_synced_at: string;
+    last_synced_at: string | null; // null until the mailbox syncs for the first time
     risk_band: string; // clean | risky | quarantine
     risk_evaluated_at?: string | null;
     warmup_health?: string; // worst warmup health_state, "" if not in a pool
@@ -735,22 +735,39 @@ export interface UserBan {
     unbanned_by_user?: AdminUserSummary | null;
 }
 
+// One user's row in user_rate_limits: the API and realtime throughput they are
+// allowed. Every value is set; a user with no row reads back the defaults.
 export interface AdminUserRateLimits {
     user_id: string;
-    limit_ws_message_pm?: number | null;
-    limit_ws_join_pm?: number | null;
-    limit_ws_event_pm?: number | null;
-    max_connections?: number | null;
-    daily_email_limit?: number | null;
-    updated_at: string;
+    limit_read_pm: number;
+    limit_write_pm: number;
+    limit_bulk_pm: number;
+    limit_unibox_pm: number;
+    limit_analytics_pm: number;
+    limit_api_calls_daily: number;
+    limit_bulk_ops_daily: number;
+    limit_ws_message_pm: number;
+    limit_ws_join_pm: number;
+    limit_ws_event_pm: number;
+    max_connections: number;
+    notes?: string | null;
+    updated_by?: string | null;
+    updated_at?: string | null; // absent when these are the defaults
 }
 
 export interface UpdateUserRateLimitsRequest {
+    limit_read_pm?: number;
+    limit_write_pm?: number;
+    limit_bulk_pm?: number;
+    limit_unibox_pm?: number;
+    limit_analytics_pm?: number;
+    limit_api_calls_daily?: number;
+    limit_bulk_ops_daily?: number;
     limit_ws_message_pm?: number;
     limit_ws_join_pm?: number;
     limit_ws_event_pm?: number;
     max_connections?: number;
-    daily_email_limit?: number;
+    notes?: string;
 }
 
 export interface BanUserRequest {

cmd/backend/main.go

@@ -599,7 +599,8 @@ func main() {
 		// and record redemptions at checkout), and the discount service audits
 		// management actions through the admin service.
 		adminRepository := repository.NewAdminRepository(primaryDB.Pool)
-		adminService = admin.NewService(adminRepository)
+		campaignLogRepository := repository.NewCampaignLogRepository(primaryDB)
+		adminService = admin.NewService(adminRepository, campaignLogRepository)
 		discountCodeRepository := repository.NewDiscountCodeRepository(primaryDB.Pool)
 		discountRedemptionRepository := repository.NewDiscountRedemptionRepository(primaryDB.Pool)
 		discountService = discount.NewService(discountCodeRepository, discountRedemptionRepository, planRepository, adminService)
@@ -692,7 +693,6 @@ func main() {
 		})
 		go webhookWorker.Run(ctx)
 		campaignProgressRepository := repository.NewCampaignProgressRepository(primaryDB.Pool)
-		campaignLogRepository := repository.NewCampaignLogRepository(primaryDB)
 		warmupService = warmupapp.NewService(warmupRepository)
 		// Fan out warmup health transitions to customer webhooks.
 		warmupService.WireWebhooks(webhookService, emailRepostory)

internal/api/handler/admin.go

@@ -3,6 +3,7 @@ package handler
 import (
 	"net/http"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/gin-gonic/gin"
@@ -217,13 +218,13 @@ func (h *Handler) AdminUpdateUserRateLimits(c *gin.Context) {
 		return
 	}
 
-	xerr := h.AdminService.UpdateUserRateLimits(c.Request.Context(), *adminID, userID, &req, c.ClientIP(), c.GetHeader("User-Agent"))
+	limits, xerr := h.AdminService.UpdateUserRateLimits(c.Request.Context(), *adminID, userID, &req, c.ClientIP(), c.GetHeader("User-Agent"))
 	if xerr != nil {
 		errx.JSON(c, xerr)
 		return
 	}
 
-	c.JSON(http.StatusOK, gin.H{"message": "rate limits updated successfully"})
+	c.JSON(http.StatusOK, limits)
 }
 
 // Worker Management Handlers
@@ -605,7 +606,18 @@ func (h *Handler) AdminStopCampaign(c *gin.Context) {
 		return
 	}
 
-	xerr := h.AdminService.StopCampaign(c.Request.Context(), *adminID, campaignID, c.ClientIP(), c.GetHeader("User-Agent"))
+	var req models.AdminStopCampaignRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		errx.JSON(c, errx.New(errx.BadRequest, "a reason is required to force-stop a campaign"))
+		return
+	}
+	reason := strings.TrimSpace(req.Reason)
+	if reason == "" {
+		errx.JSON(c, errx.New(errx.BadRequest, "a reason is required to force-stop a campaign"))
+		return
+	}
+
+	xerr := h.AdminService.StopCampaign(c.Request.Context(), *adminID, campaignID, reason, c.ClientIP(), c.GetHeader("User-Agent"))
 	if xerr != nil {
 		errx.JSON(c, xerr)
 		return

internal/app/admin/service.go

@@ -26,7 +26,7 @@ type AdminService interface {
 	GetUserCampaigns(ctx context.Context, userID uuid.UUID, cursor *uuid.UUID, limit int) (*models.AdminCampaignsResult, *errx.Error)
 	GetUserEmails(ctx context.Context, userID uuid.UUID, cursor *uuid.UUID, limit int) ([]models.AdminWorkerEmail, *models.Pagination, *errx.Error)
 	GetUserRateLimits(ctx context.Context, userID uuid.UUID) (*models.AdminUserRateLimits, *errx.Error)
-	UpdateUserRateLimits(ctx context.Context, adminID, userID uuid.UUID, update *models.UpdateUserRateLimitsRequest, ipAddress, userAgent string) *errx.Error
+	UpdateUserRateLimits(ctx context.Context, adminID, userID uuid.UUID, update *models.UpdateUserRateLimitsRequest, ipAddress, userAgent string) (*models.AdminUserRateLimits, *errx.Error)
 
 	// Worker Management
 	ListWorkers(ctx context.Context, cursor *uuid.UUID, limit int) (*models.AdminWorkersResult, *errx.Error)
@@ -51,7 +51,7 @@ type AdminService interface {
 	// Campaign Management
 	SearchCampaigns(ctx context.Context, search *models.AdminCampaignSearch) (*models.AdminCampaignsResult, *errx.Error)
 	GetCampaignDetail(ctx context.Context, campaignID uuid.UUID) (*models.AdminCampaignDetail, *errx.Error)
-	StopCampaign(ctx context.Context, adminID, campaignID uuid.UUID, ipAddress, userAgent string) *errx.Error
+	StopCampaign(ctx context.Context, adminID, campaignID uuid.UUID, reason, ipAddress, userAgent string) *errx.Error
 
 	// Analytics
 	GetPlatformOverview(ctx context.Context) (*models.PlatformOverview, *errx.Error)
@@ -91,11 +91,13 @@ type AdminService interface {
 
 type adminService struct {
 	repo repository.AdminRepository
+	// The owner-visible campaign activity feed.
+	campaignLogRepo repository.CampaignLogRepository
 }
 
 // NewService creates a new admin service
-func NewService(repo repository.AdminRepository) AdminService {
-	return &adminService{repo: repo}
+func NewService(repo repository.AdminRepository, campaignLogRepo repository.CampaignLogRepository) AdminService {
+	return &adminService{repo: repo, campaignLogRepo: campaignLogRepo}
 }
 
 // logAction logs an admin action
@@ -173,6 +175,10 @@ func (s *adminService) GetUserPreview(ctx context.Context, userID uuid.UUID) (*m
 	if preview == nil {
 		return nil, errx.ErrNotFound
 	}
+	// Same answer as GET /users/:id/rate-limits: no row means the defaults.
+	if preview.RateLimits == nil {
+		preview.RateLimits = models.DefaultAdminUserRateLimits(userID)
+	}
 	return preview, nil
 }
 
@@ -275,17 +281,21 @@ func (s *adminService) GetUserRateLimits(ctx context.Context, userID uuid.UUID)
 		sentry.CaptureException(err)
 		return nil, errx.New(errx.Internal, "failed to get rate limits")
 	}
+	// No row means the enforcement path falls back to the product defaults.
+	if limits == nil {
+		return models.DefaultAdminUserRateLimits(userID), nil
+	}
 	return limits, nil
 }
 
-func (s *adminService) UpdateUserRateLimits(ctx context.Context, adminID, userID uuid.UUID, update *models.UpdateUserRateLimitsRequest, ipAddress, userAgent string) *errx.Error {
-	if err := s.repo.UpdateUserRateLimits(ctx, userID, update); err != nil {
+func (s *adminService) UpdateUserRateLimits(ctx context.Context, adminID, userID uuid.UUID, update *models.UpdateUserRateLimitsRequest, ipAddress, userAgent string) (*models.AdminUserRateLimits, *errx.Error) {
+	if err := s.repo.UpdateUserRateLimits(ctx, userID, adminID, update); err != nil {
 		sentry.CaptureException(err)
-		return errx.New(errx.Internal, "failed to update rate limits")
+		return nil, errx.New(errx.Internal, "failed to update rate limits")
 	}
 
 	s.logAction(ctx, adminID, "update_rate_limits", "user", userID, map[string]any{"limits": update}, ipAddress, userAgent)
-	return nil
+	return s.GetUserRateLimits(ctx, userID)
 }
 
 // Worker Management
@@ -471,7 +481,7 @@ func (s *adminService) GetCampaignDetail(ctx context.Context, campaignID uuid.UU
 	return campaign, nil
 }
 
-func (s *adminService) StopCampaign(ctx context.Context, adminID, campaignID uuid.UUID, ipAddress, userAgent string) *errx.Error {
+func (s *adminService) StopCampaign(ctx context.Context, adminID, campaignID uuid.UUID, reason, ipAddress, userAgent string) *errx.Error {
 	campaign, err := s.repo.GetCampaignDetail(ctx, campaignID)
 	if err != nil {
 		sentry.CaptureException(err)
@@ -481,12 +491,33 @@ func (s *adminService) StopCampaign(ctx context.Context, adminID, campaignID uui
 		return errx.ErrNotFound
 	}
 
-	if err := s.repo.StopCampaign(ctx, campaignID); err != nil {
+	// The repository decides: a draft has never sent and a completed campaign
+	// never will, and either can become true between here and the UPDATE.
+	stopped, err := s.repo.StopCampaign(ctx, campaignID)
+	if err != nil {
 		sentry.CaptureException(err)
 		return errx.New(errx.Internal, "failed to stop campaign")
 	}
+	if !stopped {
+		return errx.New(errx.BadRequest, "campaign is not running")
+	}
 
-	s.logAction(ctx, adminID, "force_stop_campaign", "campaign", campaignID, map[string]any{"user_id": campaign.UserID}, ipAddress, userAgent)
+	// Without the reason the owner just sees a campaign that went quiet.
+	if s.campaignLogRepo != nil {
+		if err := s.campaignLogRepo.CreateLog(ctx, &repository.CampaignLogEntry{
+			CampaignID: campaignID,
+			EventType:  "stopped",
+			Message:    "Campaign force-stopped by platform staff: " + reason,
+			Metadata:   map[string]interface{}{"reason": reason},
+		}); err != nil {
+			sentry.CaptureException(err)
+		}
+	}
+
+	s.logAction(ctx, adminID, "force_stop_campaign", "campaign", campaignID, map[string]any{
+		"user_id": campaign.UserID,
+		"reason":  reason,
+	}, ipAddress, userAgent)
 	return nil
 }
 
@@ -575,6 +606,23 @@ func (s *adminService) SearchPlansForAdmin(ctx context.Context, search *models.A
 	return result, nil
 }
 
+// resolveDuration maps the API's billing period onto a durations row. An
+// unknown period is the caller's mistake, not an internal error.
+func (s *adminService) resolveDuration(ctx context.Context, d models.Duration) (*uuid.UUID, *errx.Error) {
+	if d == "" {
+		return nil, errx.New(errx.BadRequest, "duration is required")
+	}
+	id, err := s.repo.DurationIDByTitle(ctx, string(d))
+	if err != nil {
+		sentry.CaptureException(err)
+		return nil, errx.New(errx.Internal, "failed to resolve plan duration")
+	}
+	if id == nil {
+		return nil, errx.New(errx.BadRequest, "unknown plan duration")
+	}
+	return id, nil
+}
+
 func (s *adminService) CreatePlan(ctx context.Context, adminID uuid.UUID, req *models.CreatePlanRequest, ipAddress, userAgent string) (*models.Plan, *errx.Error) {
 	plan := &models.Plan{
 		ID:                 uuid.New(),
@@ -595,7 +643,12 @@ func (s *adminService) CreatePlan(ctx context.Context, adminID uuid.UUID, req *m
 		MaxEmailAccounts:   req.MaxEmailAccounts,
 	}
 
-	if err := s.repo.CreatePlan(ctx, plan); err != nil {
+	durationID, xerr := s.resolveDuration(ctx, plan.Duration)
+	if xerr != nil {
+		return nil, xerr
+	}
+
+	if err := s.repo.CreatePlan(ctx, plan, *durationID); err != nil {
 		sentry.CaptureException(err)
 		return nil, errx.New(errx.Internal, "failed to create plan")
 	}
@@ -673,7 +726,12 @@ func (s *adminService) UpdatePlan(ctx context.Context, adminID, planID uuid.UUID
 		plan.MaxEmailAccounts = req.MaxEmailAccounts
 	}
 
-	if err := s.repo.UpdatePlan(ctx, plan); err != nil {
+	durationID, xerr := s.resolveDuration(ctx, plan.Duration)
+	if xerr != nil {
+		return nil, xerr
+	}
+
+	if err := s.repo.UpdatePlan(ctx, plan, *durationID); err != nil {
 		sentry.CaptureException(err)
 		return nil, errx.New(errx.Internal, "failed to update plan")
 	}

internal/models/admin.go

@@ -179,14 +179,15 @@ type AdminUpdateWorker struct {
 // the per-mailbox health signals (risk band + worst warmup health state) so the
 // admin worker view can show how healthy the inboxes on a worker are.
 type AdminWorkerEmail struct {
-	ID              uuid.UUID  `json:"id"`
-	Email           string     `json:"email"`
-	UserID          uuid.UUID  `json:"user_id"`
-	OrganizationID  *uuid.UUID `json:"organization_id,omitempty"`
-	Status          string     `json:"status"`
-	Provider        string     `json:"provider"`
-	WarmupEnabled   bool       `json:"warmup_enabled"`
-	LastSyncedAt    time.Time  `json:"last_synced_at"`
+	ID             uuid.UUID  `json:"id"`
+	Email          string     `json:"email"`
+	UserID         uuid.UUID  `json:"user_id"`
+	OrganizationID *uuid.UUID `json:"organization_id,omitempty"`
+	Status         string     `json:"status"`
+	Provider       string     `json:"provider"`
+	WarmupEnabled  bool       `json:"warmup_enabled"`
+	// NULL until the mailbox syncs for the first time.
+	LastSyncedAt    *time.Time `json:"last_synced_at"`
 	RiskBand        string     `json:"risk_band"` // clean | risky | quarantine
 	RiskEvaluatedAt *time.Time `json:"risk_evaluated_at,omitempty"`
 	WarmupHealth    string     `json:"warmup_health,omitempty"` // worst warmup health_state, "" if not in a pool
@@ -289,6 +290,12 @@ type AdminCampaignDetail struct {
 	Organization *Organization     `json:"organization,omitempty"`
 }
 
+// AdminStopCampaignRequest carries the reason the audit trail and the owner's
+// campaign feed show, so it is required.
+type AdminStopCampaignRequest struct {
+	Reason string `json:"reason" binding:"required"`
+}
+
 // AdminCampaignsResult represents paginated campaign listing
 type AdminCampaignsResult struct {
 	Data       []AdminCampaignDetail `json:"data"`
@@ -643,24 +650,69 @@ type GrantAdminRequest struct {
 	Permissions AdminPermission `json:"permissions" binding:"required"`
 }
 
-// AdminUserRateLimits represents rate limits for a specific user
+// AdminUserRateLimits is one user's row in user_rate_limits: the API and
+// realtime throughput they are allowed. Mail volume is not a rate limit.
 type AdminUserRateLimits struct {
-	UserID           uuid.UUID `json:"user_id"`
-	LimitWSMessagePM *int      `json:"limit_ws_message_pm,omitempty"`
-	LimitWSJoinPM    *int      `json:"limit_ws_join_pm,omitempty"`
-	LimitWSEventPM   *int      `json:"limit_ws_event_pm,omitempty"`
-	MaxConnections   *int      `json:"max_connections,omitempty"`
-	DailyEmailLimit  *int      `json:"daily_email_limit,omitempty"`
-	UpdatedAt        time.Time `json:"updated_at"`
+	UserID uuid.UUID `json:"user_id"`
+
+	LimitReadPM      int `json:"limit_read_pm"`
+	LimitWritePM     int `json:"limit_write_pm"`
+	LimitBulkPM      int `json:"limit_bulk_pm"`
+	LimitUniboxPM    int `json:"limit_unibox_pm"`
+	LimitAnalyticsPM int `json:"limit_analytics_pm"`
+
+	LimitAPICallsDaily int `json:"limit_api_calls_daily"`
+	LimitBulkOpsDaily  int `json:"limit_bulk_ops_daily"`
+
+	LimitWSMessagePM int `json:"limit_ws_message_pm"`
+	LimitWSJoinPM    int `json:"limit_ws_join_pm"`
+	LimitWSEventPM   int `json:"limit_ws_event_pm"`
+	MaxConnections   int `json:"max_connections"`
+
+	Notes     *string    `json:"notes,omitempty"`
+	UpdatedBy *uuid.UUID `json:"updated_by,omitempty"`
+	// Absent when the user has no override row and these are the defaults.
+	UpdatedAt *time.Time `json:"updated_at,omitempty"`
 }
 
-// UpdateUserRateLimitsRequest represents the request to update user rate limits
+// UpdateUserRateLimitsRequest patches user_rate_limits. An omitted field is left
+// alone; there is no null state, so clearing means typing the default back in.
 type UpdateUserRateLimitsRequest struct {
+	LimitReadPM      *int `json:"limit_read_pm,omitempty"`
+	LimitWritePM     *int `json:"limit_write_pm,omitempty"`
+	LimitBulkPM      *int `json:"limit_bulk_pm,omitempty"`
+	LimitUniboxPM    *int `json:"limit_unibox_pm,omitempty"`
+	LimitAnalyticsPM *int `json:"limit_analytics_pm,omitempty"`
+
+	LimitAPICallsDaily *int `json:"limit_api_calls_daily,omitempty"`
+	LimitBulkOpsDaily  *int `json:"limit_bulk_ops_daily,omitempty"`
+
 	LimitWSMessagePM *int `json:"limit_ws_message_pm,omitempty"`
 	LimitWSJoinPM    *int `json:"limit_ws_join_pm,omitempty"`
 	LimitWSEventPM   *int `json:"limit_ws_event_pm,omitempty"`
 	MaxConnections   *int `json:"max_connections,omitempty"`
-	DailyEmailLimit  *int `json:"daily_email_limit,omitempty"`
+
+	Notes *string `json:"notes,omitempty"`
+}
+
+// DefaultAdminUserRateLimits is what the enforcement path applies to a user with
+// no override row, so the editor shows the limits actually in force.
+func DefaultAdminUserRateLimits(userID uuid.UUID) *AdminUserRateLimits {
+	d := DefaultRateLimits()
+	return &AdminUserRateLimits{
+		UserID:             userID,
+		LimitReadPM:        d.LimitReadPM,
+		LimitWritePM:       d.LimitWritePM,
+		LimitBulkPM:        d.LimitBulkPM,
+		LimitUniboxPM:      d.LimitUniboxPM,
+		LimitAnalyticsPM:   d.LimitAnalyticsPM,
+		LimitAPICallsDaily: d.LimitAPICallsDaily,
+		LimitBulkOpsDaily:  d.LimitBulkOpsDaily,
+		LimitWSMessagePM:   d.LimitWSMessagePM,
+		LimitWSJoinPM:      d.LimitWSJoinPM,
+		LimitWSEventPM:     d.LimitWSEventPM,
+		MaxConnections:     d.MaxConnections,
+	}
 }
 
 // AdminUserPreview represents a full preview of a user's account

internal/repository/admin_schema_live_test.go

@@ -0,0 +1,358 @@
+package repository
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+// Issue #209: six admin queries named schema that does not exist, so each one
+// failed on every call. The prepare sweep (TestLiveEveryQueryPrepares) proves a
+// statement CAN run; these tests prove the fixed statements do the right thing.
+//
+//	WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \
+//	  go test ./internal/repository/ -run LiveAdmin -v
+
+// adminFixture is one user with one organization, one campaign and one mailbox
+// that has never synced.
+type adminFixture struct {
+	org      uuid.UUID
+	user     uuid.UUID
+	campaign uuid.UUID
+	mailbox  uuid.UUID
+	tag      string
+}
+
+func newAdminFixture(t *testing.T, pool *pgxpool.Pool) *adminFixture {
+	t.Helper()
+	ctx := context.Background()
+	f := &adminFixture{
+		org:      uuid.New(),
+		user:     uuid.New(),
+		campaign: uuid.New(),
+		mailbox:  uuid.New(),
+	}
+	f.tag = "i209-" + f.org.String()[:8]
+
+	exec := func(sql string, args ...any) {
+		t.Helper()
+		if _, err := pool.Exec(ctx, sql, args...); err != nil {
+			t.Fatalf("fixture %q: %v", sql[:min(60, len(sql))], err)
+		}
+	}
+
+	exec(`INSERT INTO users (id, first_name, last_name, email, password_hash)
+	      VALUES ($1, 'Nadia', 'Live', $2, 'x')`, f.user, f.tag+"@test.local")
+	exec(`INSERT INTO organizations (id, name, slug, owner_user_id)
+	      VALUES ($1, 'Issue 209', $2, $3)`, f.org, f.tag, f.user)
+	exec(`INSERT INTO organization_members (organization_id, user_id, role, accepted_at)
+	      VALUES ($1, $2, 'owner', NOW())`, f.org, f.user)
+	exec(`INSERT INTO campaigns (id, user_id, organization_id, name, description, days, status, updated_at, created_at)
+	      VALUES ($1, $2, $3, 'Issue 209 outreach', '', 62, 'active', NOW(), NOW())`,
+		f.campaign, f.user, f.org)
+	// last_synced_at stays NULL: a mailbox that has been connected but never
+	// synced is the common case right after setup, and it used to break the
+	// scan for the whole list.
+	exec(`INSERT INTO email_accounts (id, user_id, organization_id, email, name, signature_plain, signature_html, provider)
+	      VALUES ($1, $2, $3, $4, 'Nadia', '', '', 'smtp_imap')`,
+		f.mailbox, f.user, f.org, f.tag+"-mb@test.local")
+
+	t.Cleanup(func() {
+		c := context.Background()
+		for _, step := range []struct {
+			sql string
+			arg any
+		}{
+			{`DELETE FROM campaign_logs WHERE campaign_id IN (SELECT id FROM campaigns WHERE organization_id = $1)`, f.org},
+			{`DELETE FROM campaigns WHERE organization_id = $1`, f.org},
+			{`DELETE FROM email_accounts WHERE organization_id = $1`, f.org},
+			{`DELETE FROM organization_members WHERE organization_id = $1`, f.org},
+			{`DELETE FROM organizations WHERE id = $1`, f.org},
+			{`DELETE FROM user_rate_limits WHERE user_id = $1`, f.user},
+			{`DELETE FROM admin_audit_logs WHERE admin_user_id = $1`, f.user},
+			{`DELETE FROM users WHERE id = $1`, f.user},
+		} {
+			if _, err := pool.Exec(c, step.sql, step.arg); err != nil {
+				t.Errorf("cleanup %q: %v", step.sql, err)
+			}
+		}
+	})
+	return f
+}
+
+// ensureDuration returns the id of the durations row with this title, creating
+// it for the duration of the test if the instance does not have one.
+func ensureDuration(t *testing.T, pool *pgxpool.Pool, title string) uuid.UUID {
+	t.Helper()
+	ctx := context.Background()
+	var id uuid.UUID
+	err := pool.QueryRow(ctx, `SELECT id FROM durations WHERE title = $1`, title).Scan(&id)
+	if err == nil {
+		return id
+	}
+	id = uuid.New()
+	if _, err := pool.Exec(ctx, `INSERT INTO durations (id, title) VALUES ($1, $2)`, id, title); err != nil {
+		t.Fatalf("create durations row %q: %v", title, err)
+	}
+	t.Cleanup(func() {
+		if _, err := pool.Exec(context.Background(), `DELETE FROM durations WHERE id = $1`, id); err != nil {
+			t.Errorf("cleanup durations row %q: %v", title, err)
+		}
+	})
+	return id
+}
+
+// The email-account section of the user preview compared a uuid column against
+// a text parameter, so the query errored and the caller swallowed it: every
+// user looked like they had no mailboxes.
+func TestLiveAdminUserPreviewListsMailboxes(t *testing.T) {
+	_, pool := liveContactDB(t)
+	f := newAdminFixture(t, pool)
+	repo := NewAdminRepository(pool)
+
+	preview, err := repo.GetUserPreview(context.Background(), f.user)
+	if err != nil {
+		t.Fatalf("GetUserPreview: %v", err)
+	}
+	if preview == nil {
+		t.Fatal("GetUserPreview returned no preview for an existing user")
+	}
+	if len(preview.EmailAccounts) != 1 || preview.EmailAccounts[0].ID != f.mailbox {
+		t.Fatalf("preview lists %d mailbox(es), want the one that was connected", len(preview.EmailAccounts))
+	}
+	if preview.EmailAccounts[0].LastSyncedAt != nil {
+		t.Fatalf("a mailbox that never synced reads last_synced_at = %v, want nil", preview.EmailAccounts[0].LastSyncedAt)
+	}
+	if len(preview.Organizations) != 1 || preview.Organizations[0].ID != f.org {
+		t.Fatalf("preview lists %d organization(s), want 1", len(preview.Organizations))
+	}
+
+	// The paginated mailbox list behind the user detail page compared the same
+	// uuid column against a text parameter, and that one surfaced as a 500.
+	emails, pagination, err := repo.GetUserEmails(context.Background(), f.user, nil, 50)
+	if err != nil {
+		t.Fatalf("GetUserEmails: %v", err)
+	}
+	if len(emails) != 1 || emails[0].ID != f.mailbox {
+		t.Fatalf("GetUserEmails returned %d mailbox(es), want the one that was connected", len(emails))
+	}
+	if pagination == nil || pagination.HasMore {
+		t.Fatalf("pagination = %+v, want a single complete page", pagination)
+	}
+}
+
+// A force-stop wrote status = 'stopped', which campaign_status has never had,
+// and stopped_at, which campaigns has never had.
+func TestLiveAdminForceStopPausesCampaign(t *testing.T) {
+	_, pool := liveContactDB(t)
+	f := newAdminFixture(t, pool)
+	repo := NewAdminRepository(pool)
+	ctx := context.Background()
+
+	stopped, err := repo.StopCampaign(ctx, f.campaign)
+	if err != nil {
+		t.Fatalf("StopCampaign: %v", err)
+	}
+	if !stopped {
+		t.Fatal("StopCampaign refused an active campaign")
+	}
+
+	var status string
+	var changedAt *time.Time
+	if err = pool.QueryRow(ctx,
+		`SELECT status::text, last_status_change_at FROM campaigns WHERE id = $1`, f.campaign,
+	).Scan(&status, &changedAt); err != nil {
+		t.Fatalf("read campaign back: %v", err)
+	}
+	if status != "paused" {
+		t.Fatalf("campaign status = %q after a force-stop, want %q", status, "paused")
+	}
+	if changedAt == nil {
+		t.Fatal("force-stop left last_status_change_at unset, so the owner's stop cooldown never starts")
+	}
+
+	// The scheduler only runs campaigns that are still 'active', which is what
+	// makes the status flip enough to halt the send loop.
+	detail, err := repo.GetCampaignDetail(ctx, f.campaign)
+	if err != nil {
+		t.Fatalf("GetCampaignDetail: %v", err)
+	}
+	if detail == nil || detail.Status != "paused" {
+		t.Fatalf("campaign detail reports %v, want a paused campaign", detail)
+	}
+}
+
+// A campaign that reaches a terminal state while the operator is deciding must
+// not be dragged back out of it, so the eligibility test lives in the UPDATE.
+func TestLiveAdminForceStopLeavesTerminalCampaignsAlone(t *testing.T) {
+	_, pool := liveContactDB(t)
+	f := newAdminFixture(t, pool)
+	repo := NewAdminRepository(pool)
+	ctx := context.Background()
+
+	for _, status := range []string{"completed", "draft"} {
+		if _, err := pool.Exec(ctx,
+			`UPDATE campaigns SET status = $2::campaign_status WHERE id = $1`, f.campaign, status,
+		); err != nil {
+			t.Fatalf("park campaign at %s: %v", status, err)
+		}
+
+		stopped, err := repo.StopCampaign(ctx, f.campaign)
+		if err != nil {
+			t.Fatalf("StopCampaign on a %s campaign: %v", status, err)
+		}
+		if stopped {
+			t.Fatalf("StopCampaign reported that it stopped a %s campaign", status)
+		}
+
+		var got string
+		if err := pool.QueryRow(ctx, `SELECT status::text FROM campaigns WHERE id = $1`, f.campaign).Scan(&got); err != nil {
+			t.Fatalf("read campaign back: %v", err)
+		}
+		if got != status {
+			t.Fatalf("a %s campaign is now %q; the stop overwrote a state it had no business touching", status, got)
+		}
+	}
+}
+
+// plans stores the billing period as duration_id (FK to durations); the admin
+// writes named a `duration` column that does not exist.
+func TestLiveAdminPlanRoundTripsDuration(t *testing.T) {
+	_, pool := liveContactDB(t)
+	repo := NewAdminRepository(pool)
+	ctx := context.Background()
+
+	// A bare install ships only the monthly duration (migration 000080); the
+	// yearly one arrives with the seed. Add whatever is missing and take it
+	// back out again, so this runs on either.
+	monthID := ensureDuration(t, pool, "month")
+	yearID := ensureDuration(t, pool, "year")
+
+	resolved, err := repo.DurationIDByTitle(ctx, "month")
+	if err != nil {
+		t.Fatalf("DurationIDByTitle(month): %v", err)
+	}
+	if resolved == nil || *resolved != monthID {
+		t.Fatalf("DurationIDByTitle(month) = %v, want %v", resolved, monthID)
+	}
+	if unknown, err := repo.DurationIDByTitle(ctx, "fortnight"); err != nil || unknown != nil {
+		t.Fatalf("DurationIDByTitle(fortnight) = %v, %v; want nil, nil so the API can answer 400", unknown, err)
+	}
+
+	name := "Issue 209 plan"
+	plan := &models.Plan{
+		ID:             uuid.New(),
+		Name:           &name,
+		MaxContacts:    1000,
+		DailyEmails:    50,
+		AccountLimit:   3,
+		Price:          49,
+		Duration:       models.DurationMonth,
+		MonthlyCredits: 25,
+	}
+	t.Cleanup(func() {
+		if _, err := pool.Exec(context.Background(), `DELETE FROM plans WHERE id = $1`, plan.ID); err != nil {
+			t.Errorf("cleanup plan: %v", err)
+		}
+	})
+
+	if err := repo.CreatePlan(ctx, plan, monthID); err != nil {
+		t.Fatalf("CreatePlan: %v", err)
+	}
+
+	got, err := repo.GetPlan(ctx, plan.ID)
+	if err != nil {
+		t.Fatalf("GetPlan: %v", err)
+	}
+	if got == nil {
+		t.Fatal("GetPlan found nothing for a plan that was just created")
+	}
+	if got.Duration != models.DurationMonth {
+		t.Fatalf("plan reads back duration %q, want %q", got.Duration, models.DurationMonth)
+	}
+	if got.MonthlyCredits != 25 {
+		t.Fatalf("plan reads back %d monthly credits, want 25", got.MonthlyCredits)
+	}
+
+	got.Duration = models.DurationYear
+	got.Price = 490
+	if err := repo.UpdatePlan(ctx, got, yearID); err != nil {
+		t.Fatalf("UpdatePlan: %v", err)
+	}
+	after, err := repo.GetPlan(ctx, plan.ID)
+	if err != nil {
+		t.Fatalf("GetPlan after update: %v", err)
+	}
+	if after.Duration != models.DurationYear {
+		t.Fatalf("plan reads back duration %q after switching to yearly, want %q", after.Duration, models.DurationYear)
+	}
+	if after.Price != 490 {
+		t.Fatalf("plan reads back price %v, want 490", after.Price)
+	}
+}
+
+// user_rate_limits has limit_api_calls_daily / limit_bulk_ops_daily and no
+// daily_email_limit, so both the read and the write failed every time.
+func TestLiveAdminUserRateLimitsRoundTrip(t *testing.T) {
+	_, pool := liveContactDB(t)
+	f := newAdminFixture(t, pool)
+	repo := NewAdminRepository(pool)
+	ctx := context.Background()
+
+	existing, err := repo.GetUserRateLimits(ctx, f.user)
+	if err != nil {
+		t.Fatalf("GetUserRateLimits before any override: %v", err)
+	}
+	if existing != nil {
+		t.Fatalf("a fresh user already has an override row: %+v", existing)
+	}
+
+	// A partial patch is the normal case, and it has to create the row without
+	// tripping the NOT NULL constraint on every column it does not mention.
+	writes := 42
+	if err := repo.UpdateUserRateLimits(ctx, f.user, f.user, &models.UpdateUserRateLimitsRequest{
+		LimitWritePM: &writes,
+	}); err != nil {
+		t.Fatalf("UpdateUserRateLimits (first, partial): %v", err)
+	}
+
+	limits, err := repo.GetUserRateLimits(ctx, f.user)
+	if err != nil {
+		t.Fatalf("GetUserRateLimits: %v", err)
+	}
+	if limits == nil {
+		t.Fatal("no override row after a successful update")
+	}
+	if limits.LimitWritePM != writes {
+		t.Fatalf("limit_write_pm = %d, want %d", limits.LimitWritePM, writes)
+	}
+	if limits.LimitReadPM == 0 || limits.LimitAPICallsDaily == 0 || limits.MaxConnections == 0 {
+		t.Fatalf("a partial patch zeroed the untouched columns: %+v", limits)
+	}
+	if limits.UpdatedBy == nil || *limits.UpdatedBy != f.user {
+		t.Fatalf("updated_by = %v, want the acting admin", limits.UpdatedBy)
+	}
+
+	// A second patch must leave the first one alone.
+	daily := 7
+	if err := repo.UpdateUserRateLimits(ctx, f.user, f.user, &models.UpdateUserRateLimitsRequest{
+		LimitBulkOpsDaily: &daily,
+	}); err != nil {
+		t.Fatalf("UpdateUserRateLimits (second, partial): %v", err)
+	}
+	after, err := repo.GetUserRateLimits(ctx, f.user)
+	if err != nil {
+		t.Fatalf("GetUserRateLimits after second patch: %v", err)
+	}
+	if after.LimitBulkOpsDaily != daily {
+		t.Fatalf("limit_bulk_ops_daily = %d, want %d", after.LimitBulkOpsDaily, daily)
+	}
+	if after.LimitWritePM != writes {
+		t.Fatalf("the second patch reset limit_write_pm to %d, want %d", after.LimitWritePM, writes)
+	}
+}

internal/repository/pg_admin.go

@@ -3,6 +3,7 @@ package repository
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -55,7 +56,7 @@ type AdminRepository interface {
 	// Campaign Management
 	SearchCampaigns(ctx context.Context, search *models.AdminCampaignSearch) (*models.AdminCampaignsResult, error)
 	GetCampaignDetail(ctx context.Context, campaignID uuid.UUID) (*models.AdminCampaignDetail, error)
-	StopCampaign(ctx context.Context, campaignID uuid.UUID) error
+	StopCampaign(ctx context.Context, campaignID uuid.UUID) (bool, error)
 
 	// Audit Logs
 	CreateAuditLog(ctx context.Context, log *models.AdminAuditLog) error
@@ -73,9 +74,10 @@ type AdminRepository interface {
 	// Plans
 	ListPlans(ctx context.Context, includePrivate bool) ([]models.Plan, error)
 	SearchPlansForAdmin(ctx context.Context, search *models.AdminPlanSearch) (*models.AdminPlansResult, error)
-	CreatePlan(ctx context.Context, plan *models.Plan) error
+	DurationIDByTitle(ctx context.Context, title string) (*uuid.UUID, error)
+	CreatePlan(ctx context.Context, plan *models.Plan, durationID uuid.UUID) error
 	GetPlan(ctx context.Context, planID uuid.UUID) (*models.Plan, error)
-	UpdatePlan(ctx context.Context, plan *models.Plan) error
+	UpdatePlan(ctx context.Context, plan *models.Plan, durationID uuid.UUID) error
 	DeletePlan(ctx context.Context, planID uuid.UUID) error
 	IsPlanInUse(ctx context.Context, planID uuid.UUID) (bool, error)
 
@@ -86,7 +88,7 @@ type AdminRepository interface {
 
 	// User Rate Limits
 	GetUserRateLimits(ctx context.Context, userID uuid.UUID) (*models.AdminUserRateLimits, error)
-	UpdateUserRateLimits(ctx context.Context, userID uuid.UUID, update *models.UpdateUserRateLimitsRequest) error
+	UpdateUserRateLimits(ctx context.Context, userID, adminID uuid.UUID, update *models.UpdateUserRateLimitsRequest) error
 
 	// Mailboxes (platform-wide). Joins email_accounts → users → orgs
 	// so the admin table can answer "whose mailbox is this and where
@@ -339,7 +341,8 @@ func (r *adminRepository) GetUserDetail(ctx context.Context, userID uuid.UUID) (
 	return &u, nil
 }
 
-// GetUserPreview gets a complete preview of a user's account
+// GetUserPreview gets a complete preview of a user's account. A section that
+// fails is an error, not an empty list: that is what hid the mailbox bug.
 func (r *adminRepository) GetUserPreview(ctx context.Context, userID uuid.UUID) (*models.AdminUserPreview, error) {
 	user, err := r.GetUserDetail(ctx, userID)
 	if err != nil {
@@ -361,14 +364,19 @@ func (r *adminRepository) GetUserPreview(ctx context.Context, userID uuid.UUID)
 		WHERE om.user_id = $1
 	`
 	orgRows, err := r.db.Query(ctx, orgQuery, userID)
-	if err == nil {
-		defer orgRows.Close()
-		for orgRows.Next() {
-			var org models.Organization
-			if err := orgRows.Scan(&org.ID, &org.Name, &org.Slug, &org.OwnerUserID, &org.CreatedAt, &org.UpdatedAt); err == nil {
-				preview.Organizations = append(preview.Organizations, org)
-			}
+	if err != nil {
+		return nil, err
+	}
+	defer orgRows.Close()
+	for orgRows.Next() {
+		var org models.Organization
+		if err := orgRows.Scan(&org.ID, &org.Name, &org.Slug, &org.OwnerUserID, &org.CreatedAt, &org.UpdatedAt); err != nil {
+			return nil, err
 		}
+		preview.Organizations = append(preview.Organizations, org)
+	}
+	if err := orgRows.Err(); err != nil {
+		return nil, err
 	}
 
 	// Get subscriptions
@@ -383,53 +391,71 @@ func (r *adminRepository) GetUserPreview(ctx context.Context, userID uuid.UUID)
 		WHERE s.user_id = $1
 	`
 	subRows, err := r.db.Query(ctx, subQuery, userID)
-	if err == nil {
-		defer subRows.Close()
-		for subRows.Next() {
-			var sub models.Subscription
-			if err := subRows.Scan(
-				&sub.ID, &sub.UserID, &sub.OrganizationID, &sub.PlanID, &sub.StripeCustomerID,
-				&sub.StripeSubscriptionID, &sub.StripePriceID, &sub.Status,
-				&sub.CurrentPeriodStart, &sub.CurrentPeriodEnd, &sub.CancelAtPeriodEnd,
-				&sub.CanceledAt, &sub.TrialStart, &sub.TrialEnd,
-				&sub.FreeTrialStartedAt, &sub.FreeTrialEndsAt, &sub.IsEnterprise,
-				&sub.CreatedAt, &sub.UpdatedAt,
-			); err == nil {
-				preview.Subscriptions = append(preview.Subscriptions, sub)
-			}
+	if err != nil {
+		return nil, err
+	}
+	defer subRows.Close()
+	for subRows.Next() {
+		var sub models.Subscription
+		if err := subRows.Scan(
+			&sub.ID, &sub.UserID, &sub.OrganizationID, &sub.PlanID, &sub.StripeCustomerID,
+			&sub.StripeSubscriptionID, &sub.StripePriceID, &sub.Status,
+			&sub.CurrentPeriodStart, &sub.CurrentPeriodEnd, &sub.CancelAtPeriodEnd,
+			&sub.CanceledAt, &sub.TrialStart, &sub.TrialEnd,
+			&sub.FreeTrialStartedAt, &sub.FreeTrialEndsAt, &sub.IsEnterprise,
+			&sub.CreatedAt, &sub.UpdatedAt,
+		); err != nil {
+			return nil, err
 		}
+		preview.Subscriptions = append(preview.Subscriptions, sub)
+	}
+	if err := subRows.Err(); err != nil {
+		return nil, err
 	}
 
-	// Get email accounts
+	// Get email accounts. user_id is uuid, not text.
 	emailQuery := `
 		SELECT id, email, user_id, organization_id, status, provider,
 			warmup IS NOT NULL as warmup_enabled, last_synced_at
 		FROM email_accounts
-		WHERE user_id = $1::text
+		WHERE user_id = $1
+		ORDER BY created_at DESC
 	`
 	emailRows, err := r.db.Query(ctx, emailQuery, userID)
-	if err == nil {
-		defer emailRows.Close()
-		for emailRows.Next() {
-			var email models.AdminWorkerEmail
-			if err := emailRows.Scan(
-				&email.ID, &email.Email, &email.UserID, &email.OrganizationID,
-				&email.Status, &email.Provider, &email.WarmupEnabled, &email.LastSyncedAt,
-			); err == nil {
-				preview.EmailAccounts = append(preview.EmailAccounts, email)
-			}
+	if err != nil {
+		return nil, err
+	}
+	defer emailRows.Close()
+	for emailRows.Next() {
+		var email models.AdminWorkerEmail
+		if err := emailRows.Scan(
+			&email.ID, &email.Email, &email.UserID, &email.OrganizationID,
+			&email.Status, &email.Provider, &email.WarmupEnabled, &email.LastSyncedAt,
+		); err != nil {
+			return nil, err
 		}
+		preview.EmailAccounts = append(preview.EmailAccounts, email)
+	}
+	if err := emailRows.Err(); err != nil {
+		return nil, err
 	}
 
 	// Get recent bans
-	bans, _ := r.GetUserBans(ctx, userID)
+	bans, err := r.GetUserBans(ctx, userID)
+	if err != nil {
+		return nil, err
+	}
 	if len(bans) > 5 {
 		bans = bans[:5]
 	}
 	preview.RecentBans = bans
 
 	// Get rate limits
-	preview.RateLimits, _ = r.GetUserRateLimits(ctx, userID)
+	limits, err := r.GetUserRateLimits(ctx, userID)
+	if err != nil {
+		return nil, err
+	}
+	preview.RateLimits = limits
 
 	return preview, nil
 }
@@ -589,14 +615,15 @@ func (r *adminRepository) GetUserEmails(ctx context.Context, userID uuid.UUID, c
 	}
 
 	args := []interface{}{userID, limit + 1}
-	whereClause := "WHERE ea.user_id = $1::text"
+	// email_accounts.user_id is uuid; casting it to text made every call fail.
+	whereClause := "WHERE ea.user_id = $1"
 	if cursor != nil {
 		whereClause += " AND ea.id < $3"
 		args = append(args, *cursor)
 	}
 
 	query := `
-		SELECT ea.id, ea.email, ea.user_id::uuid, ea.organization_id,
+		SELECT ea.id, ea.email, ea.user_id, ea.organization_id,
 			ea.status, ea.provider, ea.warmup IS NOT NULL, ea.last_synced_at
 		FROM email_accounts ea
 		` + whereClause + `
@@ -1570,13 +1597,21 @@ func (r *adminRepository) GetCampaignDetail(ctx context.Context, campaignID uuid
 	return &c, nil
 }
 
-// StopCampaign force-stops a campaign
-func (r *adminRepository) StopCampaign(ctx context.Context, campaignID uuid.UUID) error {
-	_, err := r.db.Exec(ctx, `
-		UPDATE campaigns SET status = 'stopped', stopped_at = NOW(), updated_at = NOW()
+// StopCampaign parks a running campaign at 'paused' (campaign_status has no
+// 'stopped'), and reports false when it was not in a stoppable state. The status
+// test belongs in the UPDATE: a campaign that completes concurrently must not be
+// resurrected by a stop that read it as active a moment earlier.
+func (r *adminRepository) StopCampaign(ctx context.Context, campaignID uuid.UUID) (bool, error) {
+	tag, err := r.db.Exec(ctx, `
+		UPDATE campaigns
+		SET status = 'paused', last_status_change_at = NOW(), updated_at = NOW()
 		WHERE id = $1
+		  AND status IN ('active', 'paused', 'paused_no_accounts', 'paused_trial_expired', 'paused_guardrail')
 	`, campaignID)
-	return err
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() > 0, nil
 }
 
 // CreateAuditLog creates an audit log entry
@@ -2076,22 +2111,36 @@ func (r *adminRepository) SearchPlansForAdmin(ctx context.Context, search *model
 	return result, nil
 }
 
-// CreatePlan creates a new plan
-func (r *adminRepository) CreatePlan(ctx context.Context, plan *models.Plan) error {
+// DurationIDByTitle resolves a durations.title to its id, or (nil, nil) when
+// there is no such period, so the caller can answer 400 not a NOT NULL violation.
+func (r *adminRepository) DurationIDByTitle(ctx context.Context, title string) (*uuid.UUID, error) {
+	var id uuid.UUID
+	err := r.db.QueryRow(ctx, `SELECT id FROM durations WHERE title = $1`, title).Scan(&id)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &id, nil
+}
+
+// CreatePlan creates a new plan. The billing period is duration_id, an FK.
+func (r *adminRepository) CreatePlan(ctx context.Context, plan *models.Plan, durationID uuid.UUID) error {
 	query := `
 		INSERT INTO plans (id, name, max_contacts, daily_emails, ai_generation, account_limit,
-			price, discounted_price, duration, savings, public, dedicated_workers, daily_campaign_limit,
+			price, discounted_price, duration_id, savings, public, dedicated_workers, daily_campaign_limit,
 			max_campaigns, max_active_campaigns, max_team_members, max_email_accounts,
-			created_at, updated_at)
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)
+			monthly_credits, created_at, updated_at)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)
 	`
 	now := time.Now()
 	_, err := r.db.Exec(ctx, query,
 		plan.ID, plan.Name, plan.MaxContacts, plan.DailyEmails, plan.AIGeneration, plan.AccountLimit,
-		plan.Price, plan.DiscountedPrice, plan.Duration, plan.Savings, plan.Public,
+		plan.Price, plan.DiscountedPrice, durationID, plan.Savings, plan.Public,
 		plan.DedicatedWorkers, plan.DailyCampaignLimit,
 		plan.MaxCampaigns, plan.MaxActiveCampaigns, plan.MaxTeamMembers, plan.MaxEmailAccounts,
-		now, now,
+		plan.MonthlyCredits, now, now,
 	)
 	return err
 }
@@ -2099,36 +2148,44 @@ func (r *adminRepository) CreatePlan(ctx context.Context, plan *models.Plan) err
 // GetPlan gets a plan by ID
 func (r *adminRepository) GetPlan(ctx context.Context, planID uuid.UUID) (*models.Plan, error) {
 	query := `
-		SELECT id, name, max_contacts, daily_emails, ai_generation, account_limit,
-			price, discounted_price, duration, savings, public,
-			stripe_price_id, stripe_product_id, dedicated_workers, daily_campaign_limit,
-			max_campaigns, max_active_campaigns, max_team_members, max_email_accounts,
-			updated_at, created_at
-		FROM plans WHERE id = $1
+		SELECT p.id, p.name, p.max_contacts, p.daily_emails, p.ai_generation, p.account_limit,
+			p.price, p.discounted_price, d.title, p.savings, p.public,
+			p.stripe_price_id, p.stripe_price_id_yearly, p.stripe_product_id,
+			p.dedicated_workers, p.daily_campaign_limit,
+			p.max_campaigns, p.max_active_campaigns, p.max_team_members, p.max_email_accounts,
+			p.monthly_credits, p.referral_reward_percent, p.updated_at, p.created_at
+		FROM plans p
+		LEFT JOIN durations d ON d.id = p.duration_id
+		WHERE p.id = $1
 	`
 
 	var p models.Plan
+	var duration *string
 	err := r.db.QueryRow(ctx, query, planID).Scan(
 		&p.ID, &p.Name, &p.MaxContacts, &p.DailyEmails, &p.AIGeneration, &p.AccountLimit,
-		&p.Price, &p.DiscountedPrice, &p.Duration, &p.Savings, &p.Public,
-		&p.StripePriceID, &p.StripeProductID, &p.DedicatedWorkers, &p.DailyCampaignLimit,
+		&p.Price, &p.DiscountedPrice, &duration, &p.Savings, &p.Public,
+		&p.StripePriceID, &p.StripePriceIDYearly, &p.StripeProductID,
+		&p.DedicatedWorkers, &p.DailyCampaignLimit,
 		&p.MaxCampaigns, &p.MaxActiveCampaigns, &p.MaxTeamMembers, &p.MaxEmailAccounts,
-		&p.UpdatedAt, &p.CreatedAt,
+		&p.MonthlyCredits, &p.ReferralRewardPercent, &p.UpdatedAt, &p.CreatedAt,
 	)
-	if err == pgx.ErrNoRows {
+	if errors.Is(err, pgx.ErrNoRows) {
 		return nil, nil
 	}
 	if err != nil {
 		return nil, err
 	}
+	if duration != nil {
+		p.Duration = models.Duration(*duration)
+	}
 	return &p, nil
 }
 
-// UpdatePlan updates a plan
-func (r *adminRepository) UpdatePlan(ctx context.Context, plan *models.Plan) error {
+// UpdatePlan updates a plan.
+func (r *adminRepository) UpdatePlan(ctx context.Context, plan *models.Plan, durationID uuid.UUID) error {
 	query := `
 		UPDATE plans SET name = $2, max_contacts = $3, daily_emails = $4, ai_generation = $5,
-			account_limit = $6, price = $7, discounted_price = $8, duration = $9, public = $10,
+			account_limit = $6, price = $7, discounted_price = $8, duration_id = $9, public = $10,
 			dedicated_workers = $11, daily_campaign_limit = $12, max_campaigns = $13,
 			max_active_campaigns = $14, max_team_members = $15, max_email_accounts = $16,
 			updated_at = $17
@@ -2136,7 +2193,7 @@ func (r *adminRepository) UpdatePlan(ctx context.Context, plan *models.Plan) err
 	`
 	_, err := r.db.Exec(ctx, query,
 		plan.ID, plan.Name, plan.MaxContacts, plan.DailyEmails, plan.AIGeneration,
-		plan.AccountLimit, plan.Price, plan.DiscountedPrice, plan.Duration, plan.Public,
+		plan.AccountLimit, plan.Price, plan.DiscountedPrice, durationID, plan.Public,
 		plan.DedicatedWorkers, plan.DailyCampaignLimit, plan.MaxCampaigns,
 		plan.MaxActiveCampaigns, plan.MaxTeamMembers, plan.MaxEmailAccounts,
 		time.Now(),
@@ -2409,21 +2466,27 @@ func (r *adminRepository) UpdateEnterpriseInquiry(ctx context.Context, id uuid.U
 	return err
 }
 
-// GetUserRateLimits gets rate limits for a user
+// GetUserRateLimits gets a user's API and realtime throughput limits. Mail
+// volume is not one of them: that is a per-mailbox budget and a plan entitlement.
 func (r *adminRepository) GetUserRateLimits(ctx context.Context, userID uuid.UUID) (*models.AdminUserRateLimits, error) {
 	query := `
-		SELECT user_id, limit_ws_message_pm, limit_ws_join_pm, limit_ws_event_pm,
-			max_connections, daily_email_limit, updated_at
+		SELECT user_id, limit_read_pm, limit_write_pm, limit_bulk_pm, limit_unibox_pm,
+			limit_analytics_pm, limit_api_calls_daily, limit_bulk_ops_daily,
+			limit_ws_message_pm, limit_ws_join_pm, limit_ws_event_pm, max_connections,
+			notes, updated_by, updated_at
 		FROM user_rate_limits
 		WHERE user_id = $1
 	`
 
 	var limits models.AdminUserRateLimits
 	err := r.db.QueryRow(ctx, query, userID).Scan(
-		&limits.UserID, &limits.LimitWSMessagePM, &limits.LimitWSJoinPM, &limits.LimitWSEventPM,
-		&limits.MaxConnections, &limits.DailyEmailLimit, &limits.UpdatedAt,
+		&limits.UserID, &limits.LimitReadPM, &limits.LimitWritePM, &limits.LimitBulkPM,
+		&limits.LimitUniboxPM, &limits.LimitAnalyticsPM, &limits.LimitAPICallsDaily,
+		&limits.LimitBulkOpsDaily, &limits.LimitWSMessagePM, &limits.LimitWSJoinPM,
+		&limits.LimitWSEventPM, &limits.MaxConnections,
+		&limits.Notes, &limits.UpdatedBy, &limits.UpdatedAt,
 	)
-	if err == pgx.ErrNoRows {
+	if errors.Is(err, pgx.ErrNoRows) {
 		return nil, nil
 	}
 	if err != nil {
@@ -2432,25 +2495,49 @@ func (r *adminRepository) GetUserRateLimits(ctx context.Context, userID uuid.UUI
 	return &limits, nil
 }
 
-// UpdateUserRateLimits updates rate limits for a user
-func (r *adminRepository) UpdateUserRateLimits(ctx context.Context, userID uuid.UUID, update *models.UpdateUserRateLimitsRequest) error {
-	query := `
-		INSERT INTO user_rate_limits (user_id, limit_ws_message_pm, limit_ws_join_pm, limit_ws_event_pm,
-			max_connections, daily_email_limit, updated_at)
-		VALUES ($1, $2, $3, $4, $5, $6, NOW())
-		ON CONFLICT (user_id) DO UPDATE SET
-			limit_ws_message_pm = COALESCE($2, user_rate_limits.limit_ws_message_pm),
-			limit_ws_join_pm = COALESCE($3, user_rate_limits.limit_ws_join_pm),
-			limit_ws_event_pm = COALESCE($4, user_rate_limits.limit_ws_event_pm),
-			max_connections = COALESCE($5, user_rate_limits.max_connections),
-			daily_email_limit = COALESCE($6, user_rate_limits.daily_email_limit),
+// UpdateUserRateLimits patches a user's limits. Every column is NOT NULL, so the
+// row is created first: an INSERT with the unset fields as NULL would be refused.
+func (r *adminRepository) UpdateUserRateLimits(ctx context.Context, userID, adminID uuid.UUID, update *models.UpdateUserRateLimitsRequest) error {
+	tx, err := r.db.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback(ctx)
+
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO user_rate_limits (user_id) VALUES ($1)
+		ON CONFLICT (user_id) DO NOTHING
+	`, userID); err != nil {
+		return err
+	}
+
+	if _, err := tx.Exec(ctx, `
+		UPDATE user_rate_limits SET
+			limit_read_pm = COALESCE($2, limit_read_pm),
+			limit_write_pm = COALESCE($3, limit_write_pm),
+			limit_bulk_pm = COALESCE($4, limit_bulk_pm),
+			limit_unibox_pm = COALESCE($5, limit_unibox_pm),
+			limit_analytics_pm = COALESCE($6, limit_analytics_pm),
+			limit_api_calls_daily = COALESCE($7, limit_api_calls_daily),
+			limit_bulk_ops_daily = COALESCE($8, limit_bulk_ops_daily),
+			limit_ws_message_pm = COALESCE($9, limit_ws_message_pm),
+			limit_ws_join_pm = COALESCE($10, limit_ws_join_pm),
+			limit_ws_event_pm = COALESCE($11, limit_ws_event_pm),
+			max_connections = COALESCE($12, max_connections),
+			notes = COALESCE($13, notes),
+			updated_by = $14,
 			updated_at = NOW()
-	`
-	_, err := r.db.Exec(ctx, query, userID,
-		update.LimitWSMessagePM, update.LimitWSJoinPM, update.LimitWSEventPM,
-		update.MaxConnections, update.DailyEmailLimit,
-	)
-	return err
+		WHERE user_id = $1
+	`,
+		userID, update.LimitReadPM, update.LimitWritePM, update.LimitBulkPM,
+		update.LimitUniboxPM, update.LimitAnalyticsPM, update.LimitAPICallsDaily,
+		update.LimitBulkOpsDaily, update.LimitWSMessagePM, update.LimitWSJoinPM,
+		update.LimitWSEventPM, update.MaxConnections, update.Notes, adminID,
+	); err != nil {
+		return err
+	}
+
+	return tx.Commit(ctx)
 }
 
 // SearchMailboxesForAdmin lists connected mailboxes platform-wide with

internal/repository/query_prepare_live_test.go

@@ -32,10 +32,35 @@ import (
 // ahead of time: ask the server. The same sweep found the identical defect in
 // the contacts bulk custom-field writes and in the worker install-state update.
 //
-// Deliberately narrow: only 42P08 fails the test. Other errors are reported but
-// tolerated, because the extractor below is a regex over source and will
-// occasionally pick up a fragment of a statement that is assembled at runtime.
-const indeterminateDatatype = "42P08"
+// Two classes fail the test. A parameter Postgres cannot type (42P08) is the
+// original #195 defect. A reference to schema that does not exist (issue #209 —
+// a renamed column, a table that was dropped, an enum value the type never had,
+// a comparison between types with no operator) is the same kind of bug found
+// the same way: six admin queries referenced a `plans.duration` column, a
+// `user_rate_limits.daily_email_limit` column and a `campaign_status` value
+// called 'stopped', none of which exist, and each failed 100% of the time.
+//
+// Syntax errors stay tolerated: the extractor below is a regex over source and
+// will pick up the first fragment of a statement that is assembled at runtime.
+const (
+	indeterminateDatatype = "42P08" // a $n Postgres cannot assign a type to
+	undefinedColumn       = "42703"
+	undefinedTable        = "42P01"
+	undefinedFunction     = "42883" // includes "operator does not exist"
+	undefinedObject       = "42704"
+	invalidTextRepr       = "22P02" // e.g. a literal that is not a value of an enum
+)
+
+// unrunnable is the set of SQLSTATEs that mean the statement can never execute,
+// whatever the calling code looks like.
+var unrunnable = map[string]string{
+	indeterminateDatatype: "a parameter Postgres cannot type",
+	undefinedColumn:       "a column that does not exist",
+	undefinedTable:        "a table that does not exist",
+	undefinedFunction:     "a function or operator that does not exist",
+	undefinedObject:       "an object that does not exist",
+	invalidTextRepr:       "a literal that is not a valid value of its type",
+}
 
 var (
 	// A backtick string literal in Go source.
@@ -113,7 +138,7 @@ func TestLiveEveryQueryPrepares(t *testing.T) {
 		t.Fatalf("only found %d statements to check; the extractor is broken", len(queries))
 	}
 
-	var ambiguous, other []string
+	var broken, other []string
 	for i, q := range queries {
 		_, err := conn.Prepare(ctx, fmt.Sprintf("check_%d", i), strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(q.sql), ";")))
 		if err == nil {
@@ -122,24 +147,25 @@ func TestLiveEveryQueryPrepares(t *testing.T) {
 
 		where := fmt.Sprintf("%s:%d", q.file, q.line)
 		var pgErr *pgconn.PgError
-		switch {
-		case errors.As(err, &pgErr) && pgErr.Code == indeterminateDatatype:
-			ambiguous = append(ambiguous, fmt.Sprintf("%s: %s (%s)\n\t\t%s", where, pgErr.Message, pgErr.Detail, oneLine(q.sql)))
-		default:
-			other = append(other, fmt.Sprintf("%s: %v", where, err))
+		if errors.As(err, &pgErr) {
+			if why, fatal := unrunnable[pgErr.Code]; fatal {
+				broken = append(broken, fmt.Sprintf("%s: %s -- %s (%s)\n\t\t%s",
+					where, why, pgErr.Message, pgErr.Code, oneLine(q.sql)))
+				continue
+			}
 		}
+		other = append(other, fmt.Sprintf("%s: %v", where, err))
 	}
 
-	// Informational: a fragment of a runtime-assembled statement lands here, but
-	// so does a genuinely broken query, so keep them visible.
+	// Informational: a fragment of a runtime-assembled statement lands here.
 	if len(other) > 0 {
-		t.Logf("%d statement(s) could not be prepared for other reasons (some are runtime-assembled fragments):\n%s",
+		t.Logf("%d statement(s) could not be prepared for other reasons (runtime-assembled fragments):\n%s",
 			len(other), strings.Join(other, "\n"))
 	}
 
-	if len(ambiguous) > 0 {
-		t.Fatalf("%d statement(s) have a parameter Postgres cannot type, so they can never execute:\n%s",
-			len(ambiguous), strings.Join(ambiguous, "\n"))
+	if len(broken) > 0 {
+		t.Fatalf("%d statement(s) can never execute:\n%s",
+			len(broken), strings.Join(broken, "\n"))
 	}
 
 	t.Logf("checked %d parameterised statements", len(queries))