diff --git a/admin/src/app/dashboard/UserRateLimitsDialog.tsx b/admin/src/app/dashboard/UserRateLimitsDialog.tsx
index 59019cac..6d5b3629 100644
--- a/admin/src/app/dashboard/UserRateLimitsDialog.tsx
+++ b/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
>(() => 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 +108,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 +166,10 @@ 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) : "",
- };
+// 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]: "" }),
+ {} 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..8613f370 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,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 {
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..e57a35ba 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,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")
}
diff --git a/internal/models/admin.go b/internal/models/admin.go
index 8cb4e459..e590b9be 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 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
diff --git a/internal/repository/admin_schema_live_test.go b/internal/repository/admin_schema_live_test.go
new file mode 100644
index 00000000..e7f2d2a9
--- /dev/null
+++ b/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)
+ }
+}
diff --git a/internal/repository/pg_admin.go b/internal/repository/pg_admin.go
index 4367e78b..d0a55f66 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"
@@ -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
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))