From 1c4fcff5585f23534c1bbe12354ac4ea8d5a87e8 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 27 Aug 2026 03:09:40 -0700 Subject: [PATCH 1/2] feat: make the admin panel's broken queries run: seven statements referenced schema that does not exist and failed 100% of the time, so a force-stop wrote status = 'stopped' and stopped_at to a campaign_status enum and a campaigns table that have neither, the plan writes named a duration column that became duration_id long ago, the user rate-limit read and write named a daily_email_limit column that user_rate_limits has never had, and the user preview compared email_accounts.user_id (uuid) against a text parameter and then swallowed the error so every operator saw an empty mailbox list; the same uuid = text defect in GetUserEmails, which the prepare sweep cannot see because that WHERE clause is assembled at runtime, was a live 500 on GET /admin/users/:id/emails; a stop now parks the campaign at 'paused' like the owner-facing stop and records the reason the UI has always sent and the backend has always dropped into both the audit log and the owner's campaign feed, the plan writes resolve durations.title to duration_id and answer 400 rather than a constraint violation on an unknown period, the rate-limit editor now covers the seven real limit columns instead of one that never existed and patches insert-then-update in a transaction because every column is NOT NULL, AdminWorkerEmail.LastSyncedAt is a pointer so a never-synced mailbox stops being silently dropped from every admin list, and TestLiveEveryQueryPrepares now fails on undefined columns, tables, operators and enum values instead of only reporting them --- admin/src/app/dashboard/CampaignsPage.tsx | 5 +- .../app/dashboard/UserRateLimitsDialog.tsx | 62 ++-- admin/src/lib/api/client/admin/users.ts | 3 +- admin/src/lib/api/models/admin.ts | 35 +- cmd/backend/main.go | 4 +- internal/api/handler/admin.go | 18 +- internal/app/admin/service.go | 95 +++++- internal/models/admin.go | 93 ++++- internal/repository/admin_schema_live_test.go | 321 ++++++++++++++++++ internal/repository/pg_admin.go | 270 ++++++++++----- .../repository/query_prepare_live_test.go | 58 +++- 11 files changed, 786 insertions(+), 178 deletions(-) create mode 100644 internal/repository/admin_schema_live_test.go diff --git a/admin/src/app/dashboard/CampaignsPage.tsx b/admin/src/app/dashboard/CampaignsPage.tsx index f92dae2c..ebd32bde 100644 --- a/admin/src/app/dashboard/CampaignsPage.tsx +++ b/admin/src/app/dashboard/CampaignsPage.tsx @@ -407,8 +407,9 @@ function StopCampaignDialog({ Force-stop campaign Stopping {campaign.name}. - 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.
diff --git a/admin/src/app/dashboard/UserRateLimitsDialog.tsx b/admin/src/app/dashboard/UserRateLimitsDialog.tsx index 59019cac..1ffbed29 100644 --- a/admin/src/app/dashboard/UserRateLimitsDialog.tsx +++ b/admin/src/app/dashboard/UserRateLimitsDialog.tsx @@ -1,7 +1,8 @@ -// 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: every +// column is NOT NULL with a default, so a user with no row reads back the +// product defaults and saving any field writes a full row. Leaving a field +// blank means "don't touch it" on the PATCH; there is no way to clear an +// override other than typing the default back in. import { useEffect, useState } from "react"; import { useMutation, useQueryClient } from "@tanstack/react-query"; @@ -24,14 +25,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 +65,12 @@ export function UserRateLimitsDialog({ onOpenChange: (v: boolean) => void; }) { const qc = useQueryClient(); - const [form, setForm] = useState>(() => seedForm(current)); + const [form, setForm] = useState>(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 +111,10 @@ export function UserRateLimitsDialog({ Rate limit overrides Per-user overrides for{" "} - {userEmail}. Leave blank - to keep the current value, set to 0 to - clear the override (back to plan default). + {userEmail}. Leave a + field blank to keep its current value. There is no null + state here, so 0 means zero, not + "inherit". @@ -154,20 +169,11 @@ export function UserRateLimitsDialog({ ); } -function seedForm(current: AdminUserRateLimits | null | undefined): Record { - const blank: Record = { - 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) : "", - }; +// The form starts empty on purpose: a blank field is "leave this alone", and +// the current value is shown beside the input rather than pre-filled into it. +function blankForm(): Record { + return FIELDS.reduce( + (acc, f) => ({ ...acc, [f.key]: "" }), + {} as Record, + ); } diff --git a/admin/src/lib/api/client/admin/users.ts b/admin/src/lib/api/client/admin/users.ts index efa3f17f..1e6e8d2c 100644 --- a/admin/src/lib/api/client/admin/users.ts +++ b/admin/src/lib/api/client/admin/users.ts @@ -71,7 +71,8 @@ export function unbanUser(id: string, body: UnbanUserRequest): Promise { }); } -export function getUserRateLimits(id: string): Promise { +// Always resolves: a user with no override row reads back the product defaults. +export function getUserRateLimits(id: string): Promise { return Request({ method: "GET", url: `/admin/users/${id}/rate-limits`, diff --git a/admin/src/lib/api/models/admin.ts b/admin/src/lib/api/models/admin.ts index e3841c04..1ab16872 100644 --- a/admin/src/lib/api/models/admin.ts +++ b/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,41 @@ export interface UserBan { unbanned_by_user?: AdminUserSummary | null; } +// One user's row in user_rate_limits: the API and realtime throughput this user +// is allowed. Every value is set, because every column is NOT NULL with a +// default; a user with no row gets the product defaults back. Outbound mail +// volume is not here, it is a per-mailbox budget and a plan entitlement. 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 { diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 128fbd8c..442dcadf 100644 --- a/cmd/backend/main.go +++ b/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) diff --git a/internal/api/handler/admin.go b/internal/api/handler/admin.go index 0433718b..b6536a30 100644 --- a/internal/api/handler/admin.go +++ b/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 diff --git a/internal/app/admin/service.go b/internal/app/admin/service.go index 3e4dc76e..b68386ea 100644 --- a/internal/app/admin/service.go +++ b/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,14 @@ type AdminService interface { type adminService struct { repo repository.AdminRepository + // campaignLogRepo writes to the owner-visible campaign activity feed, so an + // operator action shows up where the customer will look for it. + 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 +176,11 @@ 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 override row means the + // product defaults are what is enforced. + if preview.RateLimits == nil { + preview.RateLimits = models.DefaultAdminUserRateLimits(userID) + } return preview, nil } @@ -275,17 +283,22 @@ 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 no override, and the enforcement path falls back to the + // product defaults. Answer with those rather than nothing. + 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 +484,18 @@ 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 { +// stoppableCampaignStatuses are the statuses a force-stop can act on. A draft +// has never sent and a completed campaign never will, so stopping either is a +// no-op the operator should hear about rather than a silent success. +var stoppableCampaignStatuses = map[string]bool{ + "active": true, + "paused": true, + "paused_no_accounts": true, + "paused_trial_expired": true, + "paused_guardrail": true, +} + +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) @@ -480,13 +504,32 @@ func (s *adminService) StopCampaign(ctx context.Context, adminID, campaignID uui if campaign == nil { return errx.ErrNotFound } + if !stoppableCampaignStatuses[campaign.Status] { + return errx.New(errx.BadRequest, "campaign is not running") + } if err := s.repo.StopCampaign(ctx, campaignID); err != nil { sentry.CaptureException(err) return errx.New(errx.Internal, "failed to stop campaign") } - s.logAction(ctx, adminID, "force_stop_campaign", "campaign", campaignID, map[string]any{"user_id": campaign.UserID}, ipAddress, userAgent) + // The owner sees the stop in the campaign activity feed; without the reason + // the campaign simply appears paused with nothing to explain it. + 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 +618,24 @@ func (s *adminService) SearchPlansForAdmin(ctx context.Context, search *models.A return result, nil } +// resolveDuration maps the billing period the API speaks in ("month", "year") +// onto the durations row plans.duration_id points at. 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 +656,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 +739,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") } diff --git a/internal/models/admin.go b/internal/models/admin.go index 8cb4e459..64fc2354 100644 --- a/internal/models/admin.go +++ b/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 operator's reason for a force-stop. It +// is required: it is what the audit trail and the owner's campaign feed show. +type AdminStopCampaignRequest struct { + Reason string `json:"reason" binding:"required"` +} + // AdminCampaignsResult represents paginated campaign listing type AdminCampaignsResult struct { Data []AdminCampaignDetail `json:"data"` @@ -643,24 +650,74 @@ 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 this user is allowed. Every column is NOT NULL with a +// default, so a row that exists is a complete set of explicit overrides. +// Outbound mail volume is deliberately absent — it is a per-mailbox budget +// (email_accounts.campaign_limit) and a plan entitlement, 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 as it is; there is no null state to fall back to, so clearing an +// override means setting it back to the default value. 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 user_rate_limits row, so the admin editor shows the limits actually in +// force instead of an empty form. +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 diff --git a/internal/repository/admin_schema_live_test.go b/internal/repository/admin_schema_live_test.go new file mode 100644 index 00000000..9ae77076 --- /dev/null +++ b/internal/repository/admin_schema_live_test.go @@ -0,0 +1,321 @@ +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() + + if err := repo.StopCampaign(ctx, f.campaign); err != nil { + t.Fatalf("StopCampaign: %v", err) + } + + 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) + } +} + +// 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) + } +} diff --git a/internal/repository/pg_admin.go b/internal/repository/pg_admin.go index c43dd4e3..8deba3e1 100644 --- a/internal/repository/pg_admin.go +++ b/internal/repository/pg_admin.go @@ -3,6 +3,7 @@ package repository import ( "context" "encoding/json" + "errors" "fmt" "strconv" "strings" @@ -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,10 @@ 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. Every section is +// load-bearing for the operator reading it, so a failed section is an error +// rather than an empty list: the email-account query silently returned nothing +// for years because it compared the uuid column against a text parameter. func (r *adminRepository) GetUserPreview(ctx context.Context, userID uuid.UUID) (*models.AdminUserPreview, error) { user, err := r.GetUserDetail(ctx, userID) if err != nil { @@ -361,14 +366,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 +393,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 +617,16 @@ 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 here made every call + // fail with "operator does not exist: uuid = text" (issue #209). + 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,10 +1600,15 @@ func (r *adminRepository) GetCampaignDetail(ctx context.Context, campaignID uuid return &c, nil } -// StopCampaign force-stops a campaign +// StopCampaign force-stops a campaign. There is no "stopped" campaign_status: +// a stop parks the campaign at 'paused', exactly like the owner-facing stop, so +// the owner can inspect it and restart it once whatever tripped it is fixed. +// The scheduler cancels the parked task on its next tick because it refuses to +// run anything that is no longer 'active'. 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() + UPDATE campaigns + SET status = 'paused', last_status_change_at = NOW(), updated_at = NOW() WHERE id = $1 `, campaignID) return err @@ -2076,22 +2111,39 @@ 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 ("month", "year") to its id. +// plans.duration_id is a NOT NULL foreign key, so a plan write has to resolve +// the title the API speaks in before it can insert; an unknown title returns +// (nil, nil) so the caller can answer 400 rather than a constraint 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. plans stores the billing period as +// duration_id (FK to durations), not as a duration string. +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 +2151,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 +2196,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 +2469,29 @@ func (r *adminRepository) UpdateEnterpriseInquiry(ctx context.Context, id uuid.U return err } -// GetUserRateLimits gets rate limits for a user +// GetUserRateLimits gets rate limits for a user. There is no daily email limit +// here: user_rate_limits governs API and realtime throughput, and outbound mail +// volume is a per-mailbox budget (email_accounts.campaign_limit) plus the plan's +// daily_emails, neither of which lives in this table. 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 +2500,51 @@ 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 updates rate limits for a user. Every column is NOT NULL +// with a default, so the row is created first and then patched field by field: +// an INSERT carrying the unset fields as NULL would violate those constraints +// the first time an operator overrides a single limit. +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 diff --git a/internal/repository/query_prepare_live_test.go b/internal/repository/query_prepare_live_test.go index 0196fb04..f62df60f 100644 --- a/internal/repository/query_prepare_live_test.go +++ b/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)) From 0f715da88e5c689811fe9ce089591c8d0be7daaf Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 27 Aug 2026 03:21:25 -0700 Subject: [PATCH 2/2] feat: close the force-stop race and trim the comments the review flagged: the eligibility test moved out of the admin service and into the UPDATE's WHERE clause, so a campaign that completes between the status read and the write can no longer be dragged back to paused and recorded as force-stopped, StopCampaign reports whether it actually stopped anything and the service turns a refusal into the same 400 it used to raise from the pre-check, a new live test parks the campaign at completed and at draft and fails if either is overwritten, and the explanatory comment blocks added across pg_admin.go, the admin service, the admin models and the rate-limit dialog are cut back to the one-line form CLAUDE.md asks for --- .../app/dashboard/UserRateLimitsDialog.tsx | 10 ++-- admin/src/lib/api/models/admin.ts | 6 +-- internal/app/admin/service.go | 39 +++++--------- internal/models/admin.go | 17 +++---- internal/repository/admin_schema_live_test.go | 41 ++++++++++++++- internal/repository/pg_admin.go | 51 ++++++++----------- 6 files changed, 85 insertions(+), 79 deletions(-) diff --git a/admin/src/app/dashboard/UserRateLimitsDialog.tsx b/admin/src/app/dashboard/UserRateLimitsDialog.tsx index 1ffbed29..6d5b3629 100644 --- a/admin/src/app/dashboard/UserRateLimitsDialog.tsx +++ b/admin/src/app/dashboard/UserRateLimitsDialog.tsx @@ -1,8 +1,5 @@ -// User rate-limit override editor. user_rate_limits has no null state: every -// column is NOT NULL with a default, so a user with no row reads back the -// product defaults and saving any field writes a full row. Leaving a field -// blank means "don't touch it" on the PATCH; there is no way to clear an -// override other than typing the default back in. +// 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"; @@ -169,8 +166,7 @@ export function UserRateLimitsDialog({ ); } -// The form starts empty on purpose: a blank field is "leave this alone", and -// the current value is shown beside the input rather than pre-filled into it. +// Blank on purpose: the current value is shown beside the input, not in it. function blankForm(): Record { return FIELDS.reduce( (acc, f) => ({ ...acc, [f.key]: "" }), diff --git a/admin/src/lib/api/models/admin.ts b/admin/src/lib/api/models/admin.ts index 1ab16872..8613f370 100644 --- a/admin/src/lib/api/models/admin.ts +++ b/admin/src/lib/api/models/admin.ts @@ -735,10 +735,8 @@ export interface UserBan { unbanned_by_user?: AdminUserSummary | null; } -// One user's row in user_rate_limits: the API and realtime throughput this user -// is allowed. Every value is set, because every column is NOT NULL with a -// default; a user with no row gets the product defaults back. Outbound mail -// volume is not here, it is a per-mailbox budget and a plan entitlement. +// 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_read_pm: number; diff --git a/internal/app/admin/service.go b/internal/app/admin/service.go index b68386ea..e57a35ba 100644 --- a/internal/app/admin/service.go +++ b/internal/app/admin/service.go @@ -91,8 +91,7 @@ type AdminService interface { type adminService struct { repo repository.AdminRepository - // campaignLogRepo writes to the owner-visible campaign activity feed, so an - // operator action shows up where the customer will look for it. + // The owner-visible campaign activity feed. campaignLogRepo repository.CampaignLogRepository } @@ -176,8 +175,7 @@ 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 override row means the - // product defaults are what is enforced. + // Same answer as GET /users/:id/rate-limits: no row means the defaults. if preview.RateLimits == nil { preview.RateLimits = models.DefaultAdminUserRateLimits(userID) } @@ -283,8 +281,7 @@ 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 no override, and the enforcement path falls back to the - // product defaults. Answer with those rather than nothing. + // No row means the enforcement path falls back to the product defaults. if limits == nil { return models.DefaultAdminUserRateLimits(userID), nil } @@ -484,17 +481,6 @@ func (s *adminService) GetCampaignDetail(ctx context.Context, campaignID uuid.UU return campaign, nil } -// stoppableCampaignStatuses are the statuses a force-stop can act on. A draft -// has never sent and a completed campaign never will, so stopping either is a -// no-op the operator should hear about rather than a silent success. -var stoppableCampaignStatuses = map[string]bool{ - "active": true, - "paused": true, - "paused_no_accounts": true, - "paused_trial_expired": true, - "paused_guardrail": true, -} - 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 { @@ -504,17 +490,19 @@ func (s *adminService) StopCampaign(ctx context.Context, adminID, campaignID uui if campaign == nil { return errx.ErrNotFound } - if !stoppableCampaignStatuses[campaign.Status] { - return errx.New(errx.BadRequest, "campaign is not running") - } - 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") + } - // The owner sees the stop in the campaign activity feed; without the reason - // the campaign simply appears paused with nothing to explain it. + // 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, @@ -618,9 +606,8 @@ func (s *adminService) SearchPlansForAdmin(ctx context.Context, search *models.A return result, nil } -// resolveDuration maps the billing period the API speaks in ("month", "year") -// onto the durations row plans.duration_id points at. An unknown period is the -// caller's mistake, not an internal error. +// 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") diff --git a/internal/models/admin.go b/internal/models/admin.go index 64fc2354..e590b9be 100644 --- a/internal/models/admin.go +++ b/internal/models/admin.go @@ -290,8 +290,8 @@ type AdminCampaignDetail struct { Organization *Organization `json:"organization,omitempty"` } -// AdminStopCampaignRequest carries the operator's reason for a force-stop. It -// is required: it is what the audit trail and the owner's campaign feed show. +// 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"` } @@ -651,10 +651,7 @@ type GrantAdminRequest struct { } // AdminUserRateLimits is one user's row in user_rate_limits: the API and -// realtime throughput this user is allowed. Every column is NOT NULL with a -// default, so a row that exists is a complete set of explicit overrides. -// Outbound mail volume is deliberately absent — it is a per-mailbox budget -// (email_accounts.campaign_limit) and a plan entitlement, not a rate limit. +// realtime throughput they are allowed. Mail volume is not a rate limit. type AdminUserRateLimits struct { UserID uuid.UUID `json:"user_id"` @@ -678,9 +675,8 @@ type AdminUserRateLimits struct { UpdatedAt *time.Time `json:"updated_at,omitempty"` } -// UpdateUserRateLimitsRequest patches user_rate_limits. An omitted field is -// left as it is; there is no null state to fall back to, so clearing an -// override means setting it back to the default value. +// 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"` @@ -700,8 +696,7 @@ type UpdateUserRateLimitsRequest struct { } // DefaultAdminUserRateLimits is what the enforcement path applies to a user with -// no user_rate_limits row, so the admin editor shows the limits actually in -// force instead of an empty form. +// no override row, so the editor shows the limits actually in force. func DefaultAdminUserRateLimits(userID uuid.UUID) *AdminUserRateLimits { d := DefaultRateLimits() return &AdminUserRateLimits{ diff --git a/internal/repository/admin_schema_live_test.go b/internal/repository/admin_schema_live_test.go index 9ae77076..e7f2d2a9 100644 --- a/internal/repository/admin_schema_live_test.go +++ b/internal/repository/admin_schema_live_test.go @@ -154,13 +154,17 @@ func TestLiveAdminForceStopPausesCampaign(t *testing.T) { repo := NewAdminRepository(pool) ctx := context.Background() - if err := repo.StopCampaign(ctx, f.campaign); err != nil { + 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, + 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) @@ -183,6 +187,39 @@ func TestLiveAdminForceStopPausesCampaign(t *testing.T) { } } +// 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) { diff --git a/internal/repository/pg_admin.go b/internal/repository/pg_admin.go index 8deba3e1..0840cf62 100644 --- a/internal/repository/pg_admin.go +++ b/internal/repository/pg_admin.go @@ -56,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 @@ -341,10 +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. Every section is -// load-bearing for the operator reading it, so a failed section is an error -// rather than an empty list: the email-account query silently returned nothing -// for years because it compared the uuid column against a text parameter. +// 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 { @@ -617,8 +615,7 @@ func (r *adminRepository) GetUserEmails(ctx context.Context, userID uuid.UUID, c } args := []interface{}{userID, limit + 1} - // email_accounts.user_id is uuid. Casting it to text here made every call - // fail with "operator does not exist: uuid = text" (issue #209). + // 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" @@ -1600,18 +1597,21 @@ func (r *adminRepository) GetCampaignDetail(ctx context.Context, campaignID uuid return &c, nil } -// StopCampaign force-stops a campaign. There is no "stopped" campaign_status: -// a stop parks the campaign at 'paused', exactly like the owner-facing stop, so -// the owner can inspect it and restart it once whatever tripped it is fixed. -// The scheduler cancels the parked task on its next tick because it refuses to -// run anything that is no longer 'active'. -func (r *adminRepository) StopCampaign(ctx context.Context, campaignID uuid.UUID) error { - _, err := r.db.Exec(ctx, ` +// 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 @@ -2111,10 +2111,8 @@ func (r *adminRepository) SearchPlansForAdmin(ctx context.Context, search *model return result, nil } -// DurationIDByTitle resolves a durations.title ("month", "year") to its id. -// plans.duration_id is a NOT NULL foreign key, so a plan write has to resolve -// the title the API speaks in before it can insert; an unknown title returns -// (nil, nil) so the caller can answer 400 rather than a constraint violation. +// 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) @@ -2127,8 +2125,7 @@ func (r *adminRepository) DurationIDByTitle(ctx context.Context, title string) ( return &id, nil } -// CreatePlan creates a new plan. plans stores the billing period as -// duration_id (FK to durations), not as a duration string. +// 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, @@ -2469,10 +2466,8 @@ func (r *adminRepository) UpdateEnterpriseInquiry(ctx context.Context, id uuid.U return err } -// GetUserRateLimits gets rate limits for a user. There is no daily email limit -// here: user_rate_limits governs API and realtime throughput, and outbound mail -// volume is a per-mailbox budget (email_accounts.campaign_limit) plus the plan's -// daily_emails, neither of which lives in this table. +// 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_read_pm, limit_write_pm, limit_bulk_pm, limit_unibox_pm, @@ -2500,10 +2495,8 @@ func (r *adminRepository) GetUserRateLimits(ctx context.Context, userID uuid.UUI return &limits, nil } -// UpdateUserRateLimits updates rate limits for a user. Every column is NOT NULL -// with a default, so the row is created first and then patched field by field: -// an INSERT carrying the unset fields as NULL would violate those constraints -// the first time an operator overrides a single 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 {