diff --git a/console/src/components/preview.tsx b/console/src/components/preview.tsx
index 5d445a4e..316e3ad0 100644
--- a/console/src/components/preview.tsx
+++ b/console/src/components/preview.tsx
@@ -82,9 +82,24 @@ function EmailPreviewContent({
}
}, [data?.code?.source])
+ // In the small thumbnail (e.g. the Journey Send node) the Gmail-style
+ // header chrome doesn't fit and just looks broken — show the rendered
+ // email itself instead, clipped to the thumbnail height.
+ if (size === "small") {
+ return compiledHtml ? (
+
+
+
+ ) : (
+
+ {labels.noContent}
+
+ )
+ }
+
return (
-
+
)
}
diff --git a/console/src/oapi/management.generated.ts b/console/src/oapi/management.generated.ts
index 52ad61ba..e77573ed 100644
--- a/console/src/oapi/management.generated.ts
+++ b/console/src/oapi/management.generated.ts
@@ -4978,6 +4978,12 @@ export interface components {
results: components["schemas"]["OrganizationScheduled"][];
};
UpsertUserScheduledRequest: {
+ /**
+ * Format: uuid
+ * @description The schedule assignment instance id. Omit to create a new assignment (multiple assignments may share the same name per user); supply an existing id to update that assignment in place.
+ * @example a1b2c3d4-e5f6-7890-abcd-ef1234567890
+ */
+ id?: string;
/**
* Format: uuid
* @description The scheduled definition ID. Either scheduled_id or scheduled_name must be provided.
@@ -5018,6 +5024,12 @@ export interface components {
} | null;
};
UpsertOrganizationScheduledRequest: {
+ /**
+ * Format: uuid
+ * @description The schedule assignment instance id. Omit to create a new assignment (multiple assignments may share the same name per organization); supply an existing id to update that assignment in place.
+ * @example a1b2c3d4-e5f6-7890-abcd-ef1234567890
+ */
+ id?: string;
/**
* Format: uuid
* @description The scheduled definition ID. Either scheduled_id or scheduled_name must be provided.
diff --git a/console/src/views/campaign/NewCampaign.tsx b/console/src/views/campaign/NewCampaign.tsx
index e9340a53..96e5c332 100644
--- a/console/src/views/campaign/NewCampaign.tsx
+++ b/console/src/views/campaign/NewCampaign.tsx
@@ -105,7 +105,9 @@ export default function NewCampaign() {
)
if (!hasValidSelection) {
- form.setValue("subscription_id", filteredSubscriptions[0].id)
+ form.setValue("subscription_id", filteredSubscriptions[0].id, {
+ shouldValidate: true,
+ })
}
}, [isTransactional, subscriptionsLoading, filteredSubscriptions, subscriptionId, form])
@@ -323,7 +325,9 @@ export default function NewCampaign() {
onCheckedChange={(checked) => {
field.onChange(checked)
if (checked) {
- form.setValue("subscription_id", "")
+ form.setValue("subscription_id", "", {
+ shouldValidate: true,
+ })
}
}}
/>
diff --git a/console/src/views/journey/editor/JourneyEditor.utils.test.ts b/console/src/views/journey/editor/JourneyEditor.utils.test.ts
new file mode 100644
index 00000000..f1daae6d
--- /dev/null
+++ b/console/src/views/journey/editor/JourneyEditor.utils.test.ts
@@ -0,0 +1,49 @@
+// Unit tests for the entrance data_key defaulting: entrances need a data_key for
+// their trigger ("input") event variables to be referenceable downstream as
+// `journey..data.*`, so we assign one automatically and backfill any
+// legacy entrances that lack one.
+
+import { describe, expect, it } from "vitest"
+
+import { defaultEntranceDataKey, stepsToNodes } from "./JourneyEditor.utils"
+import type { JourneyStepMap } from "@/types"
+
+describe("defaultEntranceDataKey", () => {
+ it("uses 'entrance' when free", () => {
+ expect(defaultEntranceDataKey([])).toBe("entrance")
+ })
+
+ it("falls back to a numbered suffix when taken", () => {
+ expect(defaultEntranceDataKey(["entrance"])).toBe("entrance_2")
+ expect(defaultEntranceDataKey(["entrance", "entrance_2"])).toBe("entrance_3")
+ })
+
+ it("ignores gaps and unrelated keys", () => {
+ expect(defaultEntranceDataKey(["entrance", "purchase", "entrance_3"])).toBe("entrance_2")
+ })
+})
+
+describe("stepsToNodes entrance data_key backfill", () => {
+ it("assigns a default data_key to entrances that lack one", () => {
+ const steps: JourneyStepMap = {
+ a: { type: "entrance", name: "Entry", x: 0, y: 0 },
+ b: { type: "gate", name: "Gate", x: 0, y: 100 },
+ }
+ const { nodes } = stepsToNodes(steps, {})
+ const entrance = nodes.find((n) => n.id === "a")
+ const gate = nodes.find((n) => n.id === "b")
+ expect(entrance?.data.data_key).toBe("entrance")
+ // Non-entrance steps are left untouched.
+ expect(gate?.data.data_key).toBeUndefined()
+ })
+
+ it("preserves an existing data_key and de-duplicates against it", () => {
+ const steps: JourneyStepMap = {
+ a: { type: "entrance", name: "Entry 1", data_key: "entrance", x: 0, y: 0 },
+ b: { type: "entrance", name: "Entry 2", x: 200, y: 0 },
+ }
+ const { nodes } = stepsToNodes(steps, {})
+ expect(nodes.find((n) => n.id === "a")?.data.data_key).toBe("entrance")
+ expect(nodes.find((n) => n.id === "b")?.data.data_key).toBe("entrance_2")
+ })
+})
diff --git a/console/src/views/journey/editor/JourneyEditor.utils.ts b/console/src/views/journey/editor/JourneyEditor.utils.ts
index 9ff43a38..f11f89aa 100644
--- a/console/src/views/journey/editor/JourneyEditor.utils.ts
+++ b/console/src/views/journey/editor/JourneyEditor.utils.ts
@@ -100,6 +100,21 @@ export function createEdge({ data, sourceId, targetId, path }: CreateEdgeParams)
}
}
+/**
+ * The trigger event payload captured by an entrance is only referenceable from
+ * downstream steps (gates, campaigns, …) as `journey..data.*`, and the
+ * backend only exposes a step's state when it has a non-empty `data_key`. Pick a
+ * readable, unique default ("entrance", "entrance_2", …) so the input event
+ * variables are available without the user having to discover the data_key field.
+ */
+export function defaultEntranceDataKey(existingKeys: Iterable): string {
+ const used = new Set(existingKeys)
+ if (!used.has("entrance")) return "entrance"
+ let i = 2
+ while (used.has(`entrance_${i}`)) i++
+ return `entrance_${i}`
+}
+
export function stepsToNodes(
stepMap: JourneyStepMap,
actions: {
@@ -113,6 +128,11 @@ export function stepsToNodes(
const entries = Object.entries(stepMap)
const nodeIds = new Set(entries.map(([id]) => id))
+ // Backfill a default data_key for entrances that predate auto-assignment so
+ // their input event variables surface in downstream pickers. This only fills
+ // empty keys (non-destructive) and is persisted on the next manual save.
+ const usedDataKeys = new Set(entries.map(([, s]) => s.data_key).filter((k): k is string => !!k))
+
for (const [
id,
{ x, y, type, data, name, data_key, children, stats, stats_at, id: stepId },
@@ -120,6 +140,13 @@ export function stepsToNodes(
const { width, height, ...restData } = (data as Record) ?? {}
const sizeStyle =
typeof width === "number" && typeof height === "number" ? { width, height } : undefined
+
+ let resolvedDataKey = data_key
+ if (!resolvedDataKey && type === "entrance") {
+ resolvedDataKey = defaultEntranceDataKey(usedDataKeys)
+ usedDataKeys.add(resolvedDataKey)
+ }
+
nodes.push({
id,
position: { x, y },
@@ -128,7 +155,7 @@ export function stepsToNodes(
data: {
type,
name,
- data_key,
+ data_key: resolvedDataKey,
data: restData,
stats,
stats_at,
diff --git a/console/src/views/journey/hooks/useJourneyEditorGraph.ts b/console/src/views/journey/hooks/useJourneyEditorGraph.ts
index 758e7288..5c0eefa3 100644
--- a/console/src/views/journey/hooks/useJourneyEditorGraph.ts
+++ b/console/src/views/journey/hooks/useJourneyEditorGraph.ts
@@ -8,7 +8,12 @@ import type { EntranceTrigger } from "../components/JourneyTriggerSetup"
import type { JourneyEditorActionsValue } from "../editor/JourneyEditorActions"
import type { JourneyHintsValue } from "../editor/JourneyHints"
import type { JourneyEdge, JourneyNode } from "../editor/JourneyEditor.types"
-import { cloneNodes, getStepType, isValidJourneyConnection } from "../editor/JourneyEditor.utils"
+import {
+ cloneNodes,
+ defaultEntranceDataKey,
+ getStepType,
+ isValidJourneyConnection,
+} from "../editor/JourneyEditor.utils"
import { useJourneyFlowHandlers } from "./useJourneyFlowHandlers"
import { useKeyboardShortcuts } from "./useKeyboardShortcuts"
import { useStepEditing } from "./useStepEditing"
@@ -347,6 +352,7 @@ export function useJourneyEditorGraph({
data: {
type: "entrance",
name: t("entrance"),
+ data_key: defaultEntranceDataKey([]),
data: { ...defaultData, trigger },
editing: true,
},
diff --git a/console/src/views/journey/hooks/useJourneyFlowHandlers.ts b/console/src/views/journey/hooks/useJourneyFlowHandlers.ts
index 4b8ed080..2f21fba0 100644
--- a/console/src/views/journey/hooks/useJourneyFlowHandlers.ts
+++ b/console/src/views/journey/hooks/useJourneyFlowHandlers.ts
@@ -3,7 +3,11 @@ import { addEdge, type Connection, MarkerType, type ReactFlowInstance } from "@x
import { useTranslation } from "react-i18next"
import { createUuid } from "@/utils"
import { DATA_FORMAT, STEP_STYLE } from "./JourneyEditor.constants"
-import { getStepType, isValidJourneyConnection } from "../editor/JourneyEditor.utils"
+import {
+ defaultEntranceDataKey,
+ getStepType,
+ isValidJourneyConnection,
+} from "../editor/JourneyEditor.utils"
import type { JourneyEdge, JourneyNode } from "../editor/JourneyEditor.types"
export function useJourneyFlowHandlers(
@@ -74,6 +78,26 @@ export function useJourneyFlowHandlers(
if (entranceCount > 0) name = `${t("entrance")} ${entranceCount + 1}`
}
+ // Give entrances a default data_key so their trigger event data is
+ // immediately referenceable downstream as `journey..data.*`.
+ const data_key =
+ payload.type === "entrance"
+ ? defaultEntranceDataKey(
+ nds.map((n) => n.data.data_key).filter((k): k is string => !!k),
+ )
+ : undefined
+
+ // An exit must target an entrance (required field). Default to the
+ // first entrance so single-entrance journeys need no manual pick.
+ let stepData = data
+ if (
+ payload.type === "exit" &&
+ !(stepData as { entrance_uuid?: string }).entrance_uuid
+ ) {
+ const firstEntrance = nds.find((n) => n.data.type === "entrance")
+ if (firstEntrance) stepData = { ...stepData, entrance_uuid: firstEntrance.id }
+ }
+
const newNode: JourneyNode = {
id: createUuid(),
position: { x, y },
@@ -81,7 +105,8 @@ export function useJourneyFlowHandlers(
data: {
type: payload.type,
name,
- data,
+ data_key,
+ data: stepData,
...(isSticky ? { width: 275, height: 150 } : {}),
},
...(isSticky ? { style: { width: 275, height: 150 } } : {}),
diff --git a/console/src/views/settings/clients/store.ts b/console/src/views/settings/clients/store.ts
index beb5a1ad..12c5d2fb 100644
--- a/console/src/views/settings/clients/store.ts
+++ b/console/src/views/settings/clients/store.ts
@@ -19,7 +19,7 @@ import {
// pagination if projects start exceeding this.
export function useClients(projectId: UUID) {
const [result, , reload, loading] = useResolver(
- useCallback(() => api.authMethods.search(projectId, { limit: 200 }), [projectId]),
+ useCallback(() => api.authMethods.search(projectId, { limit: 100 }), [projectId]),
)
const clients = (result?.results ?? []).map(authMethodToClient)
return { clients, loading, reload }
diff --git a/internal/http/controllers/v1/client/oapi/resources.yml b/internal/http/controllers/v1/client/oapi/resources.yml
index 5bdf2cab..30dcd5d8 100644
--- a/internal/http/controllers/v1/client/oapi/resources.yml
+++ b/internal/http/controllers/v1/client/oapi/resources.yml
@@ -1646,6 +1646,12 @@ components:
required:
- name
properties:
+ id:
+ type: string
+ format: uuid
+ nullable: true
+ example: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
+ description: The id of the schedule assignment. Omit to create a new assignment (multiple assignments may share the same name per user); supply an existing id to update that assignment in place. The id is returned in the response.
name:
type: string
example: "renewal_date"
@@ -1685,6 +1691,12 @@ components:
- name
- identifier
properties:
+ id:
+ type: string
+ format: uuid
+ nullable: true
+ example: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
+ description: The id of the schedule assignment. Omit to create a new assignment (multiple assignments may share the same name per organization); supply an existing id to update that assignment in place. The id is returned in the response.
name:
type: string
example: "contract_renewal"
@@ -1721,12 +1733,19 @@ components:
DeleteUserScheduledRequest:
type: object
required:
- - name
+ - identifier
properties:
+ id:
+ type: string
+ format: uuid
+ nullable: true
+ example: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
+ description: The id of a specific schedule assignment to delete. When provided, only that instance is removed. Either id or name is required.
name:
type: string
+ nullable: true
example: "renewal_date"
- description: The name of the scheduled resource to delete
+ description: The name of the scheduled resource to delete. When provided (and id is omitted), every assignment with this name for the user is removed. Either id or name is required.
identifier:
$ref: "#/components/schemas/UserIdentifier"
@@ -1761,13 +1780,19 @@ components:
DeleteOrganizationScheduledRequest:
type: object
required:
- - name
- identifier
properties:
+ id:
+ type: string
+ format: uuid
+ nullable: true
+ example: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
+ description: The id of a specific schedule assignment to delete. When provided, only that instance is removed. Either id or name is required.
name:
type: string
+ nullable: true
example: "contract_renewal"
- description: The name of the scheduled resource to delete
+ description: The name of the scheduled resource to delete. When provided (and id is omitted), every assignment with this name for the organization is removed. Either id or name is required.
identifier:
$ref: "#/components/schemas/OrganizationIdentifier"
diff --git a/internal/http/controllers/v1/client/oapi/resources_gen.go b/internal/http/controllers/v1/client/oapi/resources_gen.go
index 976597ac..392e1a82 100644
--- a/internal/http/controllers/v1/client/oapi/resources_gen.go
+++ b/internal/http/controllers/v1/client/oapi/resources_gen.go
@@ -128,11 +128,14 @@ type DeleteOrganizationRequest struct {
// DeleteOrganizationScheduledRequest defines model for DeleteOrganizationScheduledRequest.
type DeleteOrganizationScheduledRequest struct {
+ // Id The id of a specific schedule assignment to delete. When provided, only that instance is removed. Either id or name is required.
+ Id *openapi_types.UUID `json:"id,omitempty"`
+
// Identifier One or more external identifiers to identify the organization
Identifier OrganizationIdentifier `json:"identifier"`
- // Name The name of the scheduled resource to delete
- Name string `json:"name"`
+ // Name The name of the scheduled resource to delete. When provided (and id is omitted), every assignment with this name for the organization is removed. Either id or name is required.
+ Name *string `json:"name,omitempty"`
}
// DeleteUserRequest defines model for DeleteUserRequest.
@@ -143,11 +146,14 @@ type DeleteUserRequest struct {
// DeleteUserScheduledRequest defines model for DeleteUserScheduledRequest.
type DeleteUserScheduledRequest struct {
+ // Id The id of a specific schedule assignment to delete. When provided, only that instance is removed. Either id or name is required.
+ Id *openapi_types.UUID `json:"id,omitempty"`
+
// Identifier One or more external identifiers to identify the user
- Identifier *UserIdentifier `json:"identifier,omitempty"`
+ Identifier UserIdentifier `json:"identifier"`
- // Name The name of the scheduled resource to delete
- Name string `json:"name"`
+ // Name The name of the scheduled resource to delete. When provided (and id is omitted), every assignment with this name for the user is removed. Either id or name is required.
+ Name *string `json:"name,omitempty"`
}
// DeviceRegistration defines model for DeviceRegistration.
@@ -448,6 +454,9 @@ type UpsertOrganizationScheduledRequest struct {
// Data Scheduled resource data
Data *map[string]any `json:"data,omitempty"`
+ // Id The id of the schedule assignment. Omit to create a new assignment (multiple assignments may share the same name per organization); supply an existing id to update that assignment in place. The id is returned in the response.
+ Id *openapi_types.UUID `json:"id,omitempty"`
+
// Identifier One or more external identifiers to identify the organization
Identifier OrganizationIdentifier `json:"identifier"`
@@ -469,6 +478,9 @@ type UpsertUserScheduledRequest struct {
// Data Scheduled resource data
Data *map[string]any `json:"data,omitempty"`
+ // Id The id of the schedule assignment. Omit to create a new assignment (multiple assignments may share the same name per user); supply an existing id to update that assignment in place. The id is returned in the response.
+ Id *openapi_types.UUID `json:"id,omitempty"`
+
// Identifier One or more external identifiers to identify the user
Identifier *UserIdentifier `json:"identifier,omitempty"`
diff --git a/internal/http/controllers/v1/client/scheduled.go b/internal/http/controllers/v1/client/scheduled.go
index 398303ff..c2835b07 100644
--- a/internal/http/controllers/v1/client/scheduled.go
+++ b/internal/http/controllers/v1/client/scheduled.go
@@ -75,17 +75,26 @@ func (srv *ScheduledController) UpsertUserScheduledClient(w http.ResponseWriter,
data = *req.Data
}
+ // Resolve the assignment id: a supplied id upserts that specific instance,
+ // otherwise mint a new one so this submission creates a fresh assignment.
+ // Multiple assignments may share the same name per user.
+ assignmentID := uuid.New()
+ if req.Id != nil {
+ assignmentID = *req.Id
+ }
+
userIDParams := auth.BoundUserIdentifiers(r.Context(), oapi.ToParams(*req.Identifier))
msg := schemas.ScheduledMsg{
- ID: uuid.New(),
- ProjectID: projectID,
- Name: req.Name,
- Type: scheduleType,
- SubjectType: "user",
- Data: data,
- Identifiers: userIDParams,
- StartAt: req.StartAt,
- Interval: req.Interval,
+ ID: uuid.New(),
+ ProjectID: projectID,
+ AssignmentID: assignmentID,
+ Name: req.Name,
+ Type: scheduleType,
+ SubjectType: "user",
+ Data: data,
+ Identifiers: userIDParams,
+ StartAt: req.StartAt,
+ Interval: req.Interval,
}
if req.ScheduledAt != nil {
@@ -99,7 +108,7 @@ func (srv *ScheduledController) UpsertUserScheduledClient(w http.ResponseWriter,
return
}
- logger.Info("user scheduled accepted for processing", zap.Stringer("id", msg.ID))
+ logger.Info("user scheduled accepted for processing", zap.Stringer("assignment_id", assignmentID))
var scheduledAt time.Time
if req.ScheduledAt != nil {
@@ -109,7 +118,7 @@ func (srv *ScheduledController) UpsertUserScheduledClient(w http.ResponseWriter,
}
json.Write(w, http.StatusAccepted, oapi.ScheduledAccepted{
- Id: msg.ID,
+ Id: assignmentID,
Name: req.Name,
ScheduledAt: scheduledAt,
Data: req.Data,
@@ -133,16 +142,16 @@ func (srv *ScheduledController) DeleteUserScheduledClient(w http.ResponseWriter,
return
}
- if req.Identifier == nil || len(*req.Identifier) == 0 {
+ if len(req.Identifier) == 0 {
srv.logger.Error("at least one identifier is required")
oapi.WriteProblem(w, problem.ErrBadRequest(problem.Describe("at least one identifier is required")))
return
}
- logger := srv.logger.With(zap.Stringer("project_id", projectID), zap.String("scheduled_name", req.Name))
+ logger := srv.logger.With(zap.Stringer("project_id", projectID))
logger.Info("deleting user scheduled")
- userID, err := srv.users.LookupUserID(ctx, projectID, auth.BoundUserIdentifiers(ctx, oapi.ToParams(*req.Identifier)))
+ userID, err := srv.users.LookupUserID(ctx, projectID, auth.BoundUserIdentifiers(ctx, oapi.ToParams(req.Identifier)))
if errors.Is(err, subjects.ErrUserNotFound) {
logger.Info("user not found")
oapi.WriteProblem(w, problem.ErrNotFound(problem.Describe("user not found")))
@@ -154,7 +163,41 @@ func (srv *ScheduledController) DeleteUserScheduledClient(w http.ResponseWriter,
return
}
- schedule, err := srv.users.GetScheduleByName(ctx, projectID, req.Name)
+ // Delete a single assignment by id when provided (after verifying it belongs
+ // to the resolved user); otherwise delete every assignment with the given name.
+ if req.Id != nil {
+ existing, err := srv.users.GetUserScheduleByID(ctx, *req.Id)
+ if errors.Is(err, sql.ErrNoRows) {
+ logger.Info("scheduled instance not found")
+ oapi.WriteProblem(w, problem.ErrNotFound(problem.Describe("scheduled instance not found")))
+ return
+ }
+ if err != nil {
+ logger.Error("failed to get user schedule by id", zap.Error(err))
+ oapi.WriteProblem(w, problem.ErrInternal())
+ return
+ }
+ if existing.UserID != userID {
+ logger.Warn("scheduled instance does not belong to user")
+ oapi.WriteProblem(w, problem.ErrNotFound(problem.Describe("scheduled instance not found")))
+ return
+ }
+ if err := srv.users.DeleteUserSchedule(ctx, *req.Id); err != nil {
+ logger.Error("failed to delete user schedule", zap.Error(err))
+ oapi.WriteProblem(w, problem.ErrInternal())
+ return
+ }
+ logger.Info("user scheduled instance deleted")
+ w.WriteHeader(http.StatusOK)
+ return
+ }
+
+ if req.Name == nil || *req.Name == "" {
+ oapi.WriteProblem(w, problem.ErrBadRequest(problem.Describe("either id or name is required")))
+ return
+ }
+
+ schedule, err := srv.users.GetScheduleByName(ctx, projectID, *req.Name)
if errors.Is(err, sql.ErrNoRows) {
logger.Info("schedule not found")
oapi.WriteProblem(w, problem.ErrNotFound(problem.Describe("schedule not found")))
@@ -166,8 +209,7 @@ func (srv *ScheduledController) DeleteUserScheduledClient(w http.ResponseWriter,
return
}
- err = srv.users.DeleteUserScheduleByScheduleID(ctx, userID, schedule.ID)
- if err != nil {
+ if err := srv.users.DeleteUserScheduleByScheduleID(ctx, userID, schedule.ID); err != nil {
logger.Error("failed to delete user schedule", zap.Error(err))
oapi.WriteProblem(w, problem.ErrInternal())
return
@@ -239,9 +281,18 @@ func (srv *ScheduledController) UpsertOrganizationScheduledClient(w http.Respons
data = *req.Data
}
+ // Resolve the assignment id: a supplied id upserts that specific instance,
+ // otherwise mint a new one so this submission creates a fresh assignment.
+ // Multiple assignments may share the same name per organization.
+ assignmentID := uuid.New()
+ if req.Id != nil {
+ assignmentID = *req.Id
+ }
+
msg := schemas.ScheduledMsg{
ID: uuid.New(),
ProjectID: projectID,
+ AssignmentID: assignmentID,
Name: req.Name,
Type: scheduleType,
SubjectType: "organization",
@@ -263,7 +314,7 @@ func (srv *ScheduledController) UpsertOrganizationScheduledClient(w http.Respons
return
}
- logger.Info("organization scheduled accepted for processing", zap.Stringer("id", msg.ID))
+ logger.Info("organization scheduled accepted for processing", zap.Stringer("assignment_id", assignmentID))
var scheduledAt time.Time
if req.ScheduledAt != nil {
@@ -273,7 +324,7 @@ func (srv *ScheduledController) UpsertOrganizationScheduledClient(w http.Respons
}
json.Write(w, http.StatusAccepted, oapi.ScheduledAccepted{
- Id: msg.ID,
+ Id: assignmentID,
Name: req.Name,
ScheduledAt: scheduledAt,
Data: req.Data,
@@ -302,7 +353,7 @@ func (srv *ScheduledController) DeleteOrganizationScheduledClient(w http.Respons
return
}
- logger := srv.logger.With(zap.Stringer("project_id", projectID), zap.String("scheduled_name", req.Name))
+ logger := srv.logger.With(zap.Stringer("project_id", projectID))
logger.Info("deleting organization scheduled")
orgID, err := srv.users.LookupOrganizationID(ctx, projectID, oapi.ToParams(req.Identifier))
@@ -317,7 +368,41 @@ func (srv *ScheduledController) DeleteOrganizationScheduledClient(w http.Respons
return
}
- schedule, err := srv.users.GetScheduleByName(ctx, projectID, req.Name)
+ // Delete a single assignment by id when provided (after verifying it belongs
+ // to the resolved organization); otherwise delete every assignment with the name.
+ if req.Id != nil {
+ existing, err := srv.users.GetOrganizationScheduleByID(ctx, *req.Id)
+ if errors.Is(err, sql.ErrNoRows) {
+ logger.Info("scheduled instance not found")
+ oapi.WriteProblem(w, problem.ErrNotFound(problem.Describe("scheduled instance not found")))
+ return
+ }
+ if err != nil {
+ logger.Error("failed to get organization schedule by id", zap.Error(err))
+ oapi.WriteProblem(w, problem.ErrInternal())
+ return
+ }
+ if existing.OrganizationID != orgID {
+ logger.Warn("scheduled instance does not belong to organization")
+ oapi.WriteProblem(w, problem.ErrNotFound(problem.Describe("scheduled instance not found")))
+ return
+ }
+ if err := srv.users.DeleteOrganizationSchedule(ctx, *req.Id); err != nil {
+ logger.Error("failed to delete organization schedule", zap.Error(err))
+ oapi.WriteProblem(w, problem.ErrInternal())
+ return
+ }
+ logger.Info("organization scheduled instance deleted")
+ w.WriteHeader(http.StatusOK)
+ return
+ }
+
+ if req.Name == nil || *req.Name == "" {
+ oapi.WriteProblem(w, problem.ErrBadRequest(problem.Describe("either id or name is required")))
+ return
+ }
+
+ schedule, err := srv.users.GetScheduleByName(ctx, projectID, *req.Name)
if errors.Is(err, sql.ErrNoRows) {
logger.Info("schedule not found")
oapi.WriteProblem(w, problem.ErrNotFound(problem.Describe("schedule not found")))
@@ -329,8 +414,7 @@ func (srv *ScheduledController) DeleteOrganizationScheduledClient(w http.Respons
return
}
- err = srv.users.DeleteOrganizationScheduleByScheduleID(ctx, orgID, schedule.ID)
- if err != nil {
+ if err := srv.users.DeleteOrganizationScheduleByScheduleID(ctx, orgID, schedule.ID); err != nil {
logger.Error("failed to delete organization schedule", zap.Error(err))
oapi.WriteProblem(w, problem.ErrInternal())
return
diff --git a/internal/http/controllers/v1/http.go b/internal/http/controllers/v1/http.go
index 5cebce76..b9d28722 100644
--- a/internal/http/controllers/v1/http.go
+++ b/internal/http/controllers/v1/http.go
@@ -89,18 +89,20 @@ func NewServer(ctx graceful.Context, logger *zap.Logger, cfg config.Node, db *st
// Mount management routes with JWT+API Key auth
mgmtoapi.HandlerWithOptions(mgmtController, mgmtoapi.ChiServerOptions{
BaseRouter: router,
+ // The generated wrapper applies these middlewares by wrapping forward
+ // (handler = mw(handler)), so the LAST entry becomes the outermost and
+ // runs FIRST. RateLimit is therefore listed before the validator so it
+ // runs *after* authentication and keys on the resolved auth method; the
+ // budget is shared with the client API, so a key cannot get a separate
+ // allowance per surface.
Middlewares: []mgmtoapi.MiddlewareFunc{
+ http.RateLimit(limiter, cfg.RateLimit.PerMinute, time.Minute, cfg.RateLimit.TrustedProxyHops, mgmtoapi.WriteProblem),
mgmtoapi.Validator(mgmtSpec, openapi3filter.Options{
AuthenticationFunc: auth.Middleware(
auth.WithJWT(cfg.Auth, mgmtStores),
auth.WithKey(mgmtStores, auth.SurfaceManagement),
),
}),
- // Runs after the validator (and thus after authentication) so the
- // limiter keys on the resolved auth method. The budget is shared
- // with the client API, so a key cannot get a separate allowance per
- // surface.
- http.RateLimit(limiter, cfg.RateLimit.PerMinute, time.Minute, cfg.RateLimit.TrustedProxyHops, mgmtoapi.WriteProblem),
},
})
@@ -110,7 +112,10 @@ func NewServer(ctx graceful.Context, logger *zap.Logger, cfg config.Node, db *st
r.Options("/api/client/*", func(w nethttp.ResponseWriter, r *nethttp.Request) {})
clientoapi.HandlerWithOptions(clientController, clientoapi.ChiServerOptions{
BaseRouter: r,
+ // Listed before the validator so it runs *after* authentication —
+ // see the management mount above for why the order is reversed.
Middlewares: []clientoapi.MiddlewareFunc{
+ http.RateLimit(limiter, cfg.RateLimit.PerMinute, time.Minute, cfg.RateLimit.TrustedProxyHops, clientoapi.WriteProblem),
clientoapi.Validator(clientSpec, openapi3filter.Options{
AuthenticationFunc: auth.Middleware(
auth.WithKey(mgmtStores, auth.SurfaceClient),
@@ -118,9 +123,6 @@ func NewServer(ctx graceful.Context, logger *zap.Logger, cfg config.Node, db *st
auth.WithTrustedIssuer(mgmtStores, jwksCache),
),
}),
- // Runs after the validator (and thus after authentication) so
- // the limiter keys on the resolved auth method.
- http.RateLimit(limiter, cfg.RateLimit.PerMinute, time.Minute, cfg.RateLimit.TrustedProxyHops, clientoapi.WriteProblem),
},
})
})
diff --git a/internal/http/controllers/v1/management/oapi/resources.yml b/internal/http/controllers/v1/management/oapi/resources.yml
index d686095c..41374bec 100644
--- a/internal/http/controllers/v1/management/oapi/resources.yml
+++ b/internal/http/controllers/v1/management/oapi/resources.yml
@@ -9930,6 +9930,11 @@ components:
UpsertUserScheduledRequest:
type: object
properties:
+ id:
+ type: string
+ format: uuid
+ description: The schedule assignment instance id. Omit to create a new assignment (multiple assignments may share the same name per user); supply an existing id to update that assignment in place.
+ example: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
scheduled_id:
type: string
format: uuid
@@ -9969,6 +9974,11 @@ components:
UpsertOrganizationScheduledRequest:
type: object
properties:
+ id:
+ type: string
+ format: uuid
+ description: The schedule assignment instance id. Omit to create a new assignment (multiple assignments may share the same name per organization); supply an existing id to update that assignment in place.
+ example: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
scheduled_id:
type: string
format: uuid
diff --git a/internal/http/controllers/v1/management/oapi/resources_gen.go b/internal/http/controllers/v1/management/oapi/resources_gen.go
index 48caa877..36f6c92c 100644
--- a/internal/http/controllers/v1/management/oapi/resources_gen.go
+++ b/internal/http/controllers/v1/management/oapi/resources_gen.go
@@ -2255,6 +2255,9 @@ type UpsertOrganizationScheduledRequest struct {
// Data Scheduled resource data
Data *json.RawMessage `json:"data,omitempty"`
+ // Id The schedule assignment instance id. Omit to create a new assignment (multiple assignments may share the same name per organization); supply an existing id to update that assignment in place.
+ Id *openapi_types.UUID `json:"id,omitempty"`
+
// Interval Interval for recurring schedules. When set, the schedule type is automatically set to recurring.
Interval *string `json:"interval,omitempty"`
@@ -2282,6 +2285,9 @@ type UpsertUserScheduledRequest struct {
// Data Scheduled resource data
Data *json.RawMessage `json:"data,omitempty"`
+ // Id The schedule assignment instance id. Omit to create a new assignment (multiple assignments may share the same name per user); supply an existing id to update that assignment in place.
+ Id *openapi_types.UUID `json:"id,omitempty"`
+
// Interval Interval for recurring schedules. When set, the schedule type is automatically set to recurring.
Interval *string `json:"interval,omitempty"`
diff --git a/internal/http/controllers/v1/management/scheduled.go b/internal/http/controllers/v1/management/scheduled.go
index 17c5f436..fb1d85f6 100644
--- a/internal/http/controllers/v1/management/scheduled.go
+++ b/internal/http/controllers/v1/management/scheduled.go
@@ -364,7 +364,19 @@ func (srv *ScheduledController) UpsertUserScheduled(w http.ResponseWriter, r *ht
body.StartAt = &now
}
- upserted, err := srv.store.UpsertUserSchedule(ctx, userID, scheduleID, body.ScheduledAt, body.StartAt, body.Interval, data)
+ // A supplied id updates that specific assignment; omitting it creates a new
+ // one (the store mints an id). Multiple assignments may share the same name.
+ var assignmentID uuid.UUID
+ if body.Id != nil {
+ assignmentID = *body.Id
+ }
+
+ upserted, err := srv.store.UpsertUserSchedule(ctx, assignmentID, userID, scheduleID, body.ScheduledAt, body.StartAt, body.Interval, data)
+ if errors.Is(err, subjects.ErrScheduleOwnershipMismatch) {
+ logger.Warn("scheduled instance id does not match user or schedule", zap.Error(err))
+ oapi.WriteProblem(w, problem.ErrBadRequest(problem.Describe("scheduled instance id does not match the user or schedule")))
+ return
+ }
if err != nil {
logger.Error("failed to upsert user schedule", zap.Error(err))
oapi.WriteProblem(w, err)
@@ -643,7 +655,19 @@ func (srv *ScheduledController) UpsertOrganizationScheduled(w http.ResponseWrite
body.StartAt = &now
}
- upserted, err := srv.store.UpsertOrganizationSchedule(ctx, organizationID, scheduleID, body.ScheduledAt, body.StartAt, body.Interval, data)
+ // A supplied id updates that specific assignment; omitting it creates a new
+ // one (the store mints an id). Multiple assignments may share the same name.
+ var assignmentID uuid.UUID
+ if body.Id != nil {
+ assignmentID = *body.Id
+ }
+
+ upserted, err := srv.store.UpsertOrganizationSchedule(ctx, assignmentID, organizationID, scheduleID, body.ScheduledAt, body.StartAt, body.Interval, data)
+ if errors.Is(err, subjects.ErrScheduleOwnershipMismatch) {
+ logger.Warn("scheduled instance id does not match organization or schedule", zap.Error(err))
+ oapi.WriteProblem(w, problem.ErrBadRequest(problem.Describe("scheduled instance id does not match the organization or schedule")))
+ return
+ }
if err != nil {
logger.Error("failed to upsert organization schedule", zap.Error(err))
oapi.WriteProblem(w, err)
diff --git a/internal/http/controllers/v1/management/scheduled_test.go b/internal/http/controllers/v1/management/scheduled_test.go
index e28c7a0a..5767c988 100644
--- a/internal/http/controllers/v1/management/scheduled_test.go
+++ b/internal/http/controllers/v1/management/scheduled_test.go
@@ -653,8 +653,10 @@ func TestUpsertUserScheduledIdempotent(t *testing.T) {
var result1 oapi.UserScheduled
require.NoError(t, json.Unmarshal(res1.Body.Bytes(), &result1))
+ // Re-upserting with the returned instance id updates that assignment in place.
time2 := time.Now().Add(72 * time.Hour).UTC().Truncate(time.Microsecond)
body2, _ := json.Marshal(oapi.UpsertUserScheduledRequest{
+ Id: &result1.Id,
ScheduledId: &sid,
ScheduledAt: &time2,
Data: &data,
@@ -669,10 +671,43 @@ func TestUpsertUserScheduledIdempotent(t *testing.T) {
var result2 oapi.UserScheduled
require.NoError(t, json.Unmarshal(res2.Body.Bytes(), &result2))
- require.Equal(t, result1.Id, result2.Id, "upsert should return same instance ID")
+ require.Equal(t, result1.Id, result2.Id, "upsert with id should return same instance ID")
require.WithinDuration(t, time2, result2.ScheduledAt, time.Second)
}
+func TestUpsertUserScheduledCreatesNewInstance(t *testing.T) {
+ t.Parallel()
+
+ tc := setupScheduledController(t)
+
+ userID := tc.createUser(t)
+ sid := tc.createSchedule(t, "user_multi", "single")
+ data := json.RawMessage(`{}`)
+
+ upsert := func(at time.Time) oapi.UserScheduled {
+ body, _ := json.Marshal(oapi.UpsertUserScheduledRequest{
+ ScheduledId: &sid,
+ ScheduledAt: &at,
+ Data: &data,
+ })
+ res := httptest.NewRecorder()
+ req := httptest.NewRequest("PUT", "/", bytes.NewReader(body))
+ req = req.WithContext(tc.actorCtx)
+ tc.controller.UpsertUserScheduled(res, req, tc.projectID, userID)
+ require.Equal(t, 200, res.Code)
+
+ var result oapi.UserScheduled
+ require.NoError(t, json.Unmarshal(res.Body.Bytes(), &result))
+ return result
+ }
+
+ // Without an id, each upsert creates a distinct assignment for the same name.
+ first := upsert(time.Now().Add(24 * time.Hour).UTC().Truncate(time.Microsecond))
+ second := upsert(time.Now().Add(48 * time.Hour).UTC().Truncate(time.Microsecond))
+
+ require.NotEqual(t, first.Id, second.Id)
+}
+
func TestDeleteUserScheduled(t *testing.T) {
t.Parallel()
@@ -1066,8 +1101,10 @@ func TestUpsertOrganizationScheduledIdempotent(t *testing.T) {
var result1 oapi.UserScheduled
require.NoError(t, json.Unmarshal(res1.Body.Bytes(), &result1))
+ // Re-upserting with the returned instance id updates that assignment in place.
time2 := time.Now().Add(72 * time.Hour).UTC().Truncate(time.Microsecond)
body2, _ := json.Marshal(oapi.UpsertOrganizationScheduledRequest{
+ Id: &result1.Id,
ScheduledId: &sid,
ScheduledAt: &time2,
Data: &data,
@@ -1082,7 +1119,7 @@ func TestUpsertOrganizationScheduledIdempotent(t *testing.T) {
var result2 oapi.UserScheduled
require.NoError(t, json.Unmarshal(res2.Body.Bytes(), &result2))
- require.Equal(t, result1.Id, result2.Id, "upsert should return same instance ID")
+ require.Equal(t, result1.Id, result2.Id, "upsert with id should return same instance ID")
require.WithinDuration(t, time2, result2.ScheduledAt, time.Second)
}
diff --git a/internal/http/ratelimit_test.go b/internal/http/ratelimit_test.go
index 595249b3..9211c91a 100644
--- a/internal/http/ratelimit_test.go
+++ b/internal/http/ratelimit_test.go
@@ -127,6 +127,73 @@ func TestRateLimitFailsOpenOnBackendError(t *testing.T) {
require.Empty(t, rec.Header().Get("Retry-After"))
}
+// authSetter mimics the validator's AuthenticationFunc: it resolves an actor
+// (here from a test header) and puts it on the request context for the inner
+// chain, exactly as auth.Middleware does via an in-place request mutation.
+func authSetter(next http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if id := r.Header.Get("X-Test-Actor"); id != "" {
+ actor := rbac.NewActor(rbac.ActorAPIKey, id)
+ r = r.WithContext(rbac.WithActor(r.Context(), actor))
+ }
+ next.ServeHTTP(w, r)
+ })
+}
+
+// TestRateLimitRunsAfterAuth guards the middleware ordering. The generated oapi
+// wrapper composes middlewares with a forward-wrapping loop
+// (handler = mw(handler)), so the LAST slice entry runs FIRST. RateLimit must
+// therefore be listed before the validator so it runs *after* authentication
+// and keys on the resolved actor. If the order regresses, the limiter runs
+// before auth, every request falls back to the shared ip: bucket, and two
+// distinct keys from the same IP starve a single budget.
+func TestRateLimitRunsAfterAuth(t *testing.T) {
+ t.Parallel()
+
+ limiter := newRedisLimiter(t)
+ rl := RateLimit(limiter, 1, time.Minute, 0, oapi.WriteProblem)
+
+ var reached int
+ final := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ reached++
+ w.WriteHeader(http.StatusOK)
+ })
+
+ // Compose exactly like the generated HandlerWithOptions wrapper: the slice
+ // is ordered [RateLimit, validator] and applied front-to-back, making the
+ // validator (authSetter) the outermost so it runs first.
+ middlewares := []func(http.Handler) http.Handler{rl, authSetter}
+ handler := http.Handler(final)
+ for _, mw := range middlewares {
+ handler = mw(handler)
+ }
+
+ // All requests share one RemoteAddr, so a buggy (pre-auth) limiter would
+ // key them all to a single ip: bucket.
+ newReq := func(actor string) *http.Request {
+ r := httptest.NewRequest(http.MethodPost, "/api/client/v1/events", nil)
+ r.RemoteAddr = "198.51.100.30:6666"
+ r.Header.Set("X-Test-Actor", actor)
+ return r
+ }
+
+ do := func(actor string) int {
+ rec := httptest.NewRecorder()
+ handler.ServeHTTP(rec, newReq(actor))
+ return rec.Code
+ }
+
+ // Actor A spends its budget of one, then is denied on the second request.
+ require.Equal(t, http.StatusOK, do("ak_A"))
+ require.Equal(t, http.StatusTooManyRequests, do("ak_A"))
+
+ // Actor B shares the same IP but must have its own budget — this only holds
+ // if the limiter keyed on the actor, i.e. it ran after auth.
+ require.Equal(t, http.StatusOK, do("ak_B"))
+
+ require.Equal(t, 2, reached, "only the two allowed requests reach the handler")
+}
+
func TestRateLimitKey(t *testing.T) {
t.Parallel()
diff --git a/internal/journeys/schedule.go b/internal/journeys/schedule.go
index 2f74244b..6cd6e5bb 100644
--- a/internal/journeys/schedule.go
+++ b/internal/journeys/schedule.go
@@ -5,6 +5,7 @@ import (
"fmt"
"time"
+ "github.com/google/uuid"
"github.com/lunogram/platform/internal/http/controllers/v1/management/oapi"
"github.com/lunogram/platform/internal/render"
"github.com/lunogram/platform/internal/store/journey"
@@ -12,6 +13,10 @@ import (
"go.uber.org/zap"
)
+// scheduleAssignmentNamespace is a fixed namespace for deriving deterministic
+// schedule-assignment ids from (journey entry, step) in journey steps.
+var scheduleAssignmentNamespace = uuid.MustParse("8e0f6a2c-3b4d-5e6f-7a8b-9c0d1e2f3a4b")
+
func HandleSchedule(ctx HandlerContext, step journey.JourneyVersionStep, state journey.JourneyUserState) (journey.JourneyUserState, journey.JourneyVersionStepChildren, error) {
config, err := DecodeStepData[oapi.ScheduleStepData](step.Data)
if err != nil {
@@ -87,7 +92,14 @@ func HandleSchedule(ctx HandlerContext, step journey.JourneyVersionStep, state j
data = json.RawMessage(rendered)
}
- userSchedule, err := scheduledStore.UpsertUserSchedule(ctx, ctx.UserID, scheduleID, scheduledAt, startAt, interval, data)
+ // Each journey entry (one pass of a user through the journey) schedules its
+ // own instance, so re-entering the journey creates an additional schedule.
+ // The id is derived deterministically from (journey entry, step) rather than
+ // minted randomly so that a redelivered/retried step message updates the same
+ // instance instead of duplicating it.
+ assignmentID := uuid.NewSHA1(scheduleAssignmentNamespace, append(state.JourneyEntryID[:], step.ID[:]...))
+
+ userSchedule, err := scheduledStore.UpsertUserSchedule(ctx, assignmentID, ctx.UserID, scheduleID, scheduledAt, startAt, interval, data)
if err != nil {
return state, nil, fmt.Errorf("failed to upsert user schedule: %w", err)
}
diff --git a/internal/journeys/schedule_test.go b/internal/journeys/schedule_test.go
index 339dc2d2..8e43032b 100644
--- a/internal/journeys/schedule_test.go
+++ b/internal/journeys/schedule_test.go
@@ -231,3 +231,61 @@ func TestHandleSchedule(t *testing.T) {
})
}
}
+
+func TestHandleScheduleMultipleEntries(t *testing.T) {
+ t.Parallel()
+
+ _, usersState, dbConn := setupStore(t)
+ pub := pubsub.NewNoopPublisher()
+ ctx := context.Background()
+ projectID := uuid.New()
+
+ userID, err := usersState.CreateUser(ctx, projectID, nil, nil, json.RawMessage(`{}`), nil, nil, []subjects.ExternalIDParam{
+ {Source: "anonymous", ExternalID: "anon_" + uuid.New().String()},
+ })
+ require.NoError(t, err)
+
+ step := journey.JourneyVersionStep{
+ ID: uuid.New(),
+ Type: ScheduleStepType,
+ Data: json.RawMessage(`{"schedule_name":"reentry_reminder","scheduled_at":"2030-01-01T00:00:00Z"}`),
+ Children: []journey.JourneyVersionStepChild{
+ {ChildExternalID: "next-step"},
+ },
+ }
+
+ run := func(entryID uuid.UUID) {
+ hctx := HandlerContext{
+ Context: ctx,
+ DB: dbConn,
+ Publisher: pub,
+ ProjectID: projectID,
+ UserID: userID,
+ Data: map[string]any{},
+ }
+ _, _, runErr := HandleSchedule(hctx, step, journey.JourneyUserState{JourneyEntryID: entryID})
+ require.NoError(t, runErr)
+ }
+
+ scheduledStore := subjects.NewScheduledStore(dbConn, zap.NewNop())
+
+ entry1 := uuid.New()
+ run(entry1)
+ // A redelivered/retried step message within the same journey entry must not
+ // duplicate the assignment.
+ run(entry1)
+
+ schedule, err := scheduledStore.GetScheduleByName(ctx, projectID, "reentry_reminder")
+ require.NoError(t, err)
+
+ afterFirst, err := scheduledStore.ListUserSchedulesByScheduleID(ctx, userID, schedule.ID)
+ require.NoError(t, err)
+ require.Len(t, afterFirst, 1, "same journey entry must not duplicate the assignment")
+
+ // A second journey entry schedules an additional instance with the same name.
+ run(uuid.New())
+
+ afterSecond, err := scheduledStore.ListUserSchedulesByScheduleID(ctx, userID, schedule.ID)
+ require.NoError(t, err)
+ require.Len(t, afterSecond, 2, "a new journey entry must create an additional assignment")
+}
diff --git a/internal/pubsub/consumer/scheduled.go b/internal/pubsub/consumer/scheduled.go
index 2b092beb..bf3cf1cf 100644
--- a/internal/pubsub/consumer/scheduled.go
+++ b/internal/pubsub/consumer/scheduled.go
@@ -3,6 +3,7 @@ package consumer
import (
"context"
"encoding/json"
+ "errors"
"fmt"
"time"
@@ -18,6 +19,22 @@ import (
"go.uber.org/zap"
)
+// scheduledAssignmentNamespace derives a stable assignment id for scheduled
+// messages that omit assignment_id, keyed by (subject, schedule). Producers that
+// do not set an id (e.g. the user/organization anniversary schedules) thereby
+// stay idempotent under at-least-once redelivery — a redelivered message upserts
+// the same assignment instead of minting a fresh one and duplicating it.
+var scheduledAssignmentNamespace = uuid.MustParse("c1d2e3f4-5a6b-7c8d-9e0f-1a2b3c4d5e6f")
+
+// resolveAssignmentID returns the message's assignment id, or a deterministic id
+// derived from (subjectID, scheduleID) when the message omits one.
+func resolveAssignmentID(provided, subjectID, scheduleID uuid.UUID) uuid.UUID {
+ if provided != uuid.Nil {
+ return provided
+ }
+ return uuid.NewSHA1(scheduledAssignmentNamespace, append(subjectID[:], scheduleID[:]...))
+}
+
// resolveUserID ensures scheduled.UserID is populated, looking it up by
// identifiers when necessary.
func resolveUserID(ctx context.Context, logger *zap.Logger, txState *subjects.State, scheduled *schemas.ScheduledMsg) error {
@@ -122,7 +139,12 @@ func ScheduledHandler(logger *zap.Logger, db *sqlx.DB, usrs *subjects.State, pub
if err := resolveUserID(ctx, logger, txState, &scheduled); err != nil {
return err
}
- if _, err := txState.UpsertUserSchedule(ctx, scheduled.UserID, scheduled.ScheduledID, scheduledAt, scheduled.StartAt, scheduled.Interval, data); err != nil {
+ assignmentID := resolveAssignmentID(scheduled.AssignmentID, scheduled.UserID, scheduled.ScheduledID)
+ if _, err := txState.UpsertUserSchedule(ctx, assignmentID, scheduled.UserID, scheduled.ScheduledID, scheduledAt, scheduled.StartAt, scheduled.Interval, data); err != nil {
+ if errors.Is(err, subjects.ErrScheduleOwnershipMismatch) {
+ logger.Error("user schedule assignment id does not match subject/schedule", zap.Error(err))
+ return Permanent(err)
+ }
logger.Error("failed to upsert user schedule", zap.Error(err))
return err
}
@@ -131,7 +153,12 @@ func ScheduledHandler(logger *zap.Logger, db *sqlx.DB, usrs *subjects.State, pub
if err := resolveOrganizationID(ctx, logger, txState, &scheduled); err != nil {
return err
}
- if _, err := txState.UpsertOrganizationSchedule(ctx, scheduled.OrganizationID, scheduled.ScheduledID, scheduledAt, scheduled.StartAt, scheduled.Interval, data); err != nil {
+ assignmentID := resolveAssignmentID(scheduled.AssignmentID, scheduled.OrganizationID, scheduled.ScheduledID)
+ if _, err := txState.UpsertOrganizationSchedule(ctx, assignmentID, scheduled.OrganizationID, scheduled.ScheduledID, scheduledAt, scheduled.StartAt, scheduled.Interval, data); err != nil {
+ if errors.Is(err, subjects.ErrScheduleOwnershipMismatch) {
+ logger.Error("organization schedule assignment id does not match subject/schedule", zap.Error(err))
+ return Permanent(err)
+ }
logger.Error("failed to upsert organization schedule", zap.Error(err))
return err
}
diff --git a/internal/pubsub/consumer/scheduled_test.go b/internal/pubsub/consumer/scheduled_test.go
new file mode 100644
index 00000000..30dadcae
--- /dev/null
+++ b/internal/pubsub/consumer/scheduled_test.go
@@ -0,0 +1,31 @@
+package consumer
+
+import (
+ "testing"
+
+ "github.com/google/uuid"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestResolveAssignmentID(t *testing.T) {
+ t.Parallel()
+
+ subject := uuid.New()
+ schedule := uuid.New()
+
+ // A provided id is passed through unchanged.
+ provided := uuid.New()
+ assert.Equal(t, provided, resolveAssignmentID(provided, subject, schedule))
+
+ // An omitted id is derived deterministically from (subject, schedule), so a
+ // redelivered message resolves to the same assignment and stays idempotent.
+ first := resolveAssignmentID(uuid.Nil, subject, schedule)
+ second := resolveAssignmentID(uuid.Nil, subject, schedule)
+ require.NotEqual(t, uuid.Nil, first)
+ assert.Equal(t, first, second)
+
+ // Different subjects or schedules derive different ids.
+ assert.NotEqual(t, first, resolveAssignmentID(uuid.Nil, uuid.New(), schedule))
+ assert.NotEqual(t, first, resolveAssignmentID(uuid.Nil, subject, uuid.New()))
+}
diff --git a/internal/pubsub/schemas/events.go b/internal/pubsub/schemas/events.go
index a3a671c1..2b38de06 100644
--- a/internal/pubsub/schemas/events.go
+++ b/internal/pubsub/schemas/events.go
@@ -686,6 +686,7 @@ type ScheduledMsg struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
ScheduledID uuid.UUID `json:"scheduled_id"`
+ AssignmentID uuid.UUID `json:"assignment_id,omitempty"` // the per-subject assignment (user_schedules/organization_schedules) id; zero means create a new assignment
Name string `json:"name"`
Type string `json:"type"` // "single" or "recurring"
SubjectType string `json:"subject_type"` // "user" or "organization"
diff --git a/internal/store/subjects/migrations/1784500000_next_schedule_occurrence.down.sql b/internal/store/subjects/migrations/1784500000_next_schedule_occurrence.down.sql
new file mode 100644
index 00000000..a4069ae9
--- /dev/null
+++ b/internal/store/subjects/migrations/1784500000_next_schedule_occurrence.down.sql
@@ -0,0 +1 @@
+DROP FUNCTION IF EXISTS next_schedule_occurrence(TIMESTAMPTZ, INTERVAL, BIGINT, TIMESTAMPTZ);
diff --git a/internal/store/subjects/migrations/1784500000_next_schedule_occurrence.up.sql b/internal/store/subjects/migrations/1784500000_next_schedule_occurrence.up.sql
new file mode 100644
index 00000000..a6bd8f01
--- /dev/null
+++ b/internal/store/subjects/migrations/1784500000_next_schedule_occurrence.up.sql
@@ -0,0 +1,63 @@
+-- next_schedule_occurrence returns the next occurrence of a recurring schedule:
+-- the smallest N > current_occurrence for which (anchor + N * step) is strictly
+-- after as_of, together with that N so it can be persisted. The timestamp is
+-- always multiplied from the anchor (anchor + N * step) rather than chained
+-- (prev + step), which avoids the drift that calendar intervals such as
+-- '1 month' would otherwise accumulate via month-end clamping.
+--
+-- It replaces the previous generate_series(1, 10000) scan, which both capped how
+-- far ahead a schedule could be advanced (a schedule dormant for more than 10000
+-- intervals could never resume) and was O(N) in the number of skipped occurrences.
+--
+-- Accuracy: the result is EXACT, not approximate. The epoch-based estimate only
+-- chooses where the correction loops start -- it never enters the result. The two
+-- loops below settle on N purely via exact interval arithmetic (anchor + n * step),
+-- so the returned N satisfies, for any strictly-increasing (positive) interval:
+-- (1) anchor + N * step > as_of (in the future)
+-- (2) anchor + (N - 1) * step <= as_of (it is the FIRST such occurrence)
+-- unless N = current_occurrence + 1
+-- (3) N >= current_occurrence + 1 (never moves backwards)
+-- This is the same triple the old scan satisfied, so behaviour is identical where
+-- the old scan was in range and well-defined beyond it. The epoch estimate lands
+-- close to N, so the loops do little work, and there is no cap on the gap.
+CREATE OR REPLACE FUNCTION next_schedule_occurrence(
+ anchor TIMESTAMPTZ,
+ step INTERVAL,
+ current_occurrence BIGINT,
+ as_of TIMESTAMPTZ DEFAULT NOW()
+) RETURNS TABLE(next_at TIMESTAMPTZ, occurrence BIGINT)
+LANGUAGE plpgsql
+STABLE
+AS $$
+DECLARE
+ n BIGINT;
+ step_secs DOUBLE PRECISION := EXTRACT(EPOCH FROM step);
+BEGIN
+ IF step_secs <= 0 THEN
+ RAISE EXCEPTION 'schedule interval must be positive, got %', step;
+ END IF;
+
+ -- O(1) starting estimate from the average interval length. This only seeds
+ -- the correction loops below; the exact N comes from them, not from here.
+ n := GREATEST(
+ current_occurrence + 1,
+ FLOOR(EXTRACT(EPOCH FROM (as_of - anchor)) / step_secs)::BIGINT
+ );
+
+ -- Correct exactly using interval arithmetic only. First walk back while the
+ -- previous occurrence is still in the future (estimate overshot), enforcing
+ -- property (3) via the n > current_occurrence + 1 guard. Then walk forward
+ -- while the current occurrence is not yet past as_of (estimate undershot).
+ -- On exit, anchor + (n-1)*step <= as_of < anchor + n*step, i.e. properties
+ -- (1) and (2) hold. Both loops run only a few iterations.
+ WHILE n > current_occurrence + 1 AND anchor + (n - 1) * step > as_of LOOP
+ n := n - 1;
+ END LOOP;
+
+ WHILE anchor + n * step <= as_of LOOP
+ n := n + 1;
+ END LOOP;
+
+ RETURN QUERY SELECT anchor + n * step, n;
+END;
+$$;
diff --git a/internal/store/subjects/migrations/1784600000_schedule_multi_assignment.down.sql b/internal/store/subjects/migrations/1784600000_schedule_multi_assignment.down.sql
new file mode 100644
index 00000000..8db8dcd3
--- /dev/null
+++ b/internal/store/subjects/migrations/1784600000_schedule_multi_assignment.down.sql
@@ -0,0 +1,9 @@
+-- Restore the single-assignment-per-name constraint. Note: this will fail if any
+-- subject already holds multiple assignments for the same schedule definition;
+-- such duplicates must be reconciled before rolling back.
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_user_schedules_user_schedule_unique
+ ON user_schedules (user_id, schedule_id);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_organization_schedules_org_schedule_unique
+ ON organization_schedules (organization_id, schedule_id);
diff --git a/internal/store/subjects/migrations/1784600000_schedule_multi_assignment.up.sql b/internal/store/subjects/migrations/1784600000_schedule_multi_assignment.up.sql
new file mode 100644
index 00000000..6e841e9e
--- /dev/null
+++ b/internal/store/subjects/migrations/1784600000_schedule_multi_assignment.up.sql
@@ -0,0 +1,10 @@
+-- Allow multiple schedule assignments with the same name per subject.
+-- Previously a (user_id, schedule_id) / (organization_id, schedule_id) pair was
+-- unique, so a subject could hold at most one assignment per named schedule and a
+-- second submit overwrote the first. Assignments are now addressed by their own
+-- primary key (id): submitting an existing id upserts that row, while omitting it
+-- creates a new instance. The non-unique lookup indexes (idx_user_schedules_user,
+-- idx_organization_schedules_org) are retained for query performance.
+
+DROP INDEX IF EXISTS idx_user_schedules_user_schedule_unique;
+DROP INDEX IF EXISTS idx_organization_schedules_org_schedule_unique;
diff --git a/internal/store/subjects/scheduled.go b/internal/store/subjects/scheduled.go
index 46545df0..f8425c82 100644
--- a/internal/store/subjects/scheduled.go
+++ b/internal/store/subjects/scheduled.go
@@ -3,6 +3,7 @@ package subjects
import (
"context"
"encoding/json"
+ "errors"
"fmt"
"time"
@@ -13,6 +14,13 @@ import (
"go.uber.org/zap"
)
+// ErrScheduleOwnershipMismatch is returned when an upsert targets an existing
+// schedule-assignment id that belongs to a different subject or schedule
+// definition. Assignments are addressed by their own id, but the id must
+// resolve to a row for the same (subject, schedule) pair; otherwise the upsert
+// is rejected rather than silently rewriting another subject's assignment.
+var ErrScheduleOwnershipMismatch = errors.New("schedule assignment id belongs to a different subject or schedule")
+
func (s *ScheduledStore) ValidateInterval(ctx context.Context, interval string) bool {
var positive bool
err := s.db.GetContext(ctx, &positive, `SELECT $1::interval > '0 seconds'::interval`, interval)
@@ -487,10 +495,20 @@ func (s *ScheduledStore) UpdateUserSchedule(ctx context.Context, id uuid.UUID, s
return &us, nil
}
-// UpsertUserSchedule inserts a new user schedule or updates the existing one
-// for the given (userID, scheduleID) pair using ON CONFLICT. After the upsert
-// it recomputes pending scheduled events.
-func (s *ScheduledStore) UpsertUserSchedule(ctx context.Context, userID, scheduleID uuid.UUID, scheduledAt *time.Time, startAt *time.Time, interval *string, data json.RawMessage) (*UserSchedule, error) {
+// UpsertUserSchedule creates or updates a user schedule assignment addressed by
+// its own id: a nil id (or one not yet stored) inserts a new assignment, while an
+// existing id updates that assignment in place. Assignments are keyed by id
+// rather than (userID, scheduleID), so a user may hold multiple assignments for
+// the same schedule. A supplied id must resolve to a row for the same
+// (userID, scheduleID); otherwise ErrScheduleOwnershipMismatch is returned. After
+// the upsert it recomputes pending scheduled events.
+func (s *ScheduledStore) UpsertUserSchedule(ctx context.Context, id, userID, scheduleID uuid.UUID, scheduledAt *time.Time, startAt *time.Time, interval *string, data json.RawMessage) (*UserSchedule, error) {
+ // A nil id means "create a new assignment"; mint one so the row is
+ // addressable by the caller afterwards.
+ if id == uuid.Nil {
+ id = uuid.New()
+ }
+
// For recurring schedules, anchor_at starts equal to start_at.
anchorAt := startAt
@@ -506,20 +524,28 @@ func (s *ScheduledStore) UpsertUserSchedule(ctx context.Context, userID, schedul
occurrence = result.Occurrence
}
+ // Assignments are keyed by their own id: a new id inserts a fresh instance,
+ // an existing id updates it. The ON CONFLICT guard ensures a supplied id
+ // cannot be used to overwrite a different subject's or schedule's assignment.
stmt := `
- INSERT INTO user_schedules (user_id, schedule_id, scheduled_at, start_at, anchor_at, interval, occurrence, data)
- VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
- ON CONFLICT (user_id, schedule_id) DO UPDATE SET
+ INSERT INTO user_schedules (id, user_id, schedule_id, scheduled_at, start_at, anchor_at, interval, occurrence, data)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+ ON CONFLICT (id) DO UPDATE SET
scheduled_at = COALESCE(EXCLUDED.scheduled_at, user_schedules.scheduled_at),
start_at = COALESCE(EXCLUDED.start_at, user_schedules.start_at),
anchor_at = COALESCE(EXCLUDED.anchor_at, user_schedules.anchor_at),
interval = COALESCE(EXCLUDED.interval, user_schedules.interval),
occurrence = COALESCE(EXCLUDED.occurrence, user_schedules.occurrence),
data = COALESCE(EXCLUDED.data, user_schedules.data)
+ WHERE user_schedules.user_id = EXCLUDED.user_id AND user_schedules.schedule_id = EXCLUDED.schedule_id
RETURNING id, user_id, schedule_id, scheduled_at, start_at, anchor_at, interval, occurrence, COALESCE(data, '{}'::jsonb) AS data, paused_at, created_at, updated_at`
var us UserSchedule
- err := s.db.GetContext(ctx, &us, stmt, userID, scheduleID, scheduledAt, startAt, anchorAt, interval, occurrence, data)
+ err := s.db.GetContext(ctx, &us, stmt, id, userID, scheduleID, scheduledAt, startAt, anchorAt, interval, occurrence, data)
+ if errors.Is(err, store.ErrNoRows) {
+ // The id exists but its (user, schedule) did not satisfy the guard.
+ return nil, ErrScheduleOwnershipMismatch
+ }
if err != nil {
return nil, err
}
@@ -681,7 +707,7 @@ func (s *ScheduledStore) ListUserSchedules(ctx context.Context, projectID, userI
us.occurrence, COALESCE(us.data, '{}'::jsonb) AS data, us.paused_at, us.created_at, us.updated_at,
EXISTS (
SELECT 1 FROM user_scheduled_events use
- WHERE use.user_schedule_id = us.id AND use.fired_at IS NULL
+ WHERE use.user_schedule_id = us.id AND use.fired_at IS NULL AND use.fire_at <= NOW()
) AS has_pending_events,
COUNT(*) OVER () AS total_count
FROM user_schedules us
@@ -820,14 +846,7 @@ type nextOccurrenceResult struct {
// such that anchor + N * interval > NOW(), and returns both the timestamp and
// the occurrence number N so it can be persisted to avoid drift.
func (s *ScheduledStore) computeNextOccurrence(ctx context.Context, interval string, anchor time.Time) (nextOccurrenceResult, error) {
- stmt := `
- SELECT
- $1::timestamptz + n * $2::interval AS next_at,
- n AS occurrence
- FROM generate_series(1, 10000) AS n
- WHERE $1::timestamptz + n * $2::interval > NOW()
- ORDER BY n
- LIMIT 1`
+ stmt := `SELECT next_at, occurrence FROM next_schedule_occurrence($1::timestamptz, $2::interval, 0)`
var result nextOccurrenceResult
err := s.db.GetContext(ctx, &result, stmt, anchor, interval)
@@ -918,19 +937,14 @@ func (s *ScheduledStore) AdvanceAndGenerateUserScheduleEvents(ctx context.Contex
return nil
}
- // Atomically increment occurrence and compute scheduled_at = anchor_at + occurrence * interval.
- // The new occurrence is the smallest N > current occurrence where anchor_at + N * interval > NOW().
+ // Atomically advance occurrence and scheduled_at to the next future occurrence
+ // (anchor_at + N * interval, smallest N > current occurrence). See the
+ // next_schedule_occurrence migration: it has no upper bound on the catch-up gap.
advanceStmt := `
UPDATE user_schedules
- SET occurrence = sub.n,
- scheduled_at = sub.next_at
- FROM (
- SELECT n, $2::timestamptz + n * $3::interval AS next_at
- FROM generate_series($4::int + 1, $4::int + 10000) AS n
- WHERE $2::timestamptz + n * $3::interval > NOW()
- ORDER BY n
- LIMIT 1
- ) sub
+ SET occurrence = f.occurrence,
+ scheduled_at = f.next_at
+ FROM next_schedule_occurrence($2::timestamptz, $3::interval, $4::bigint) f
WHERE id = $1
RETURNING user_schedules.scheduled_at, user_schedules.occurrence`
@@ -1094,10 +1108,21 @@ func (s *ScheduledStore) UpdateOrganizationSchedule(ctx context.Context, id uuid
return &os, nil
}
-// UpsertOrganizationSchedule inserts a new organization schedule or updates the
-// existing one for the given (organizationID, scheduleID) pair using ON CONFLICT.
-// After the upsert it recomputes pending scheduled events.
-func (s *ScheduledStore) UpsertOrganizationSchedule(ctx context.Context, organizationID, scheduleID uuid.UUID, scheduledAt *time.Time, startAt *time.Time, interval *string, data json.RawMessage) (*OrganizationSchedule, error) {
+// UpsertOrganizationSchedule creates or updates an organization schedule
+// assignment addressed by its own id: a nil id (or one not yet stored) inserts a
+// new assignment, while an existing id updates that assignment in place.
+// Assignments are keyed by id rather than (organizationID, scheduleID), so an
+// organization may hold multiple assignments for the same schedule. A supplied id
+// must resolve to a row for the same (organizationID, scheduleID); otherwise
+// ErrScheduleOwnershipMismatch is returned. After the upsert it recomputes
+// pending scheduled events.
+func (s *ScheduledStore) UpsertOrganizationSchedule(ctx context.Context, id, organizationID, scheduleID uuid.UUID, scheduledAt *time.Time, startAt *time.Time, interval *string, data json.RawMessage) (*OrganizationSchedule, error) {
+ // A nil id means "create a new assignment"; mint one so the row is
+ // addressable by the caller afterwards.
+ if id == uuid.Nil {
+ id = uuid.New()
+ }
+
// For recurring schedules, anchor_at starts equal to start_at.
anchorAt := startAt
@@ -1113,20 +1138,28 @@ func (s *ScheduledStore) UpsertOrganizationSchedule(ctx context.Context, organiz
occurrence = result.Occurrence
}
+ // Assignments are keyed by their own id: a new id inserts a fresh instance,
+ // an existing id updates it. The ON CONFLICT guard ensures a supplied id
+ // cannot be used to overwrite a different subject's or schedule's assignment.
stmt := `
- INSERT INTO organization_schedules (organization_id, schedule_id, scheduled_at, start_at, anchor_at, interval, occurrence, data)
- VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
- ON CONFLICT (organization_id, schedule_id) DO UPDATE SET
+ INSERT INTO organization_schedules (id, organization_id, schedule_id, scheduled_at, start_at, anchor_at, interval, occurrence, data)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+ ON CONFLICT (id) DO UPDATE SET
scheduled_at = COALESCE(EXCLUDED.scheduled_at, organization_schedules.scheduled_at),
start_at = COALESCE(EXCLUDED.start_at, organization_schedules.start_at),
anchor_at = COALESCE(EXCLUDED.anchor_at, organization_schedules.anchor_at),
interval = COALESCE(EXCLUDED.interval, organization_schedules.interval),
occurrence = COALESCE(EXCLUDED.occurrence, organization_schedules.occurrence),
data = COALESCE(EXCLUDED.data, organization_schedules.data)
+ WHERE organization_schedules.organization_id = EXCLUDED.organization_id AND organization_schedules.schedule_id = EXCLUDED.schedule_id
RETURNING id, organization_id, schedule_id, scheduled_at, start_at, anchor_at, interval, occurrence, COALESCE(data, '{}'::jsonb) AS data, paused_at, created_at, updated_at`
var os OrganizationSchedule
- err := s.db.GetContext(ctx, &os, stmt, organizationID, scheduleID, scheduledAt, startAt, anchorAt, interval, occurrence, data)
+ err := s.db.GetContext(ctx, &os, stmt, id, organizationID, scheduleID, scheduledAt, startAt, anchorAt, interval, occurrence, data)
+ if errors.Is(err, store.ErrNoRows) {
+ // The id exists but its (organization, schedule) did not satisfy the guard.
+ return nil, ErrScheduleOwnershipMismatch
+ }
if err != nil {
return nil, err
}
@@ -1288,7 +1321,7 @@ func (s *ScheduledStore) ListOrganizationSchedules(ctx context.Context, projectI
os.occurrence, COALESCE(os.data, '{}'::jsonb) AS data, os.paused_at, os.created_at, os.updated_at,
EXISTS (
SELECT 1 FROM organization_scheduled_events ose
- WHERE ose.organization_schedule_id = os.id AND ose.fired_at IS NULL
+ WHERE ose.organization_schedule_id = os.id AND ose.fired_at IS NULL AND ose.fire_at <= NOW()
) AS has_pending_events,
COUNT(*) OVER () AS total_count
FROM organization_schedules os
@@ -1570,19 +1603,14 @@ func (s *ScheduledStore) AdvanceAndGenerateOrgScheduleEvents(ctx context.Context
return nil
}
- // Atomically increment occurrence and compute scheduled_at = anchor_at + occurrence * interval.
- // The new occurrence is the smallest N > current occurrence where anchor_at + N * interval > NOW().
+ // Atomically advance occurrence and scheduled_at to the next future occurrence
+ // (anchor_at + N * interval, smallest N > current occurrence). See the
+ // next_schedule_occurrence migration: it has no upper bound on the catch-up gap.
advanceStmt := `
UPDATE organization_schedules
- SET occurrence = sub.n,
- scheduled_at = sub.next_at
- FROM (
- SELECT n, $2::timestamptz + n * $3::interval AS next_at
- FROM generate_series($4::int + 1, $4::int + 10000) AS n
- WHERE $2::timestamptz + n * $3::interval > NOW()
- ORDER BY n
- LIMIT 1
- ) sub
+ SET occurrence = f.occurrence,
+ scheduled_at = f.next_at
+ FROM next_schedule_occurrence($2::timestamptz, $3::interval, $4::bigint) f
WHERE id = $1
RETURNING organization_schedules.scheduled_at, organization_schedules.occurrence`
diff --git a/internal/store/subjects/scheduled_test.go b/internal/store/subjects/scheduled_test.go
index e9658ddc..3d636ce6 100644
--- a/internal/store/subjects/scheduled_test.go
+++ b/internal/store/subjects/scheduled_test.go
@@ -589,18 +589,68 @@ func TestUpsertUserSchedule(t *testing.T) {
require.NoError(t, err)
time1 := time.Now().Add(24 * time.Hour).UTC().Truncate(time.Microsecond)
- us1, err := db.UpsertUserSchedule(ctx, userID, scheduleID, &time1, nil, nil, json.RawMessage(`{"v":1}`))
+ us1, err := db.UpsertUserSchedule(ctx, uuid.Nil, userID, scheduleID, &time1, nil, nil, json.RawMessage(`{"v":1}`))
require.NoError(t, err)
+ // Re-upserting with the same assignment id updates that row in place.
time2 := time.Now().Add(48 * time.Hour).UTC().Truncate(time.Microsecond)
- us2, err := db.UpsertUserSchedule(ctx, userID, scheduleID, &time2, nil, nil, json.RawMessage(`{"v":2}`))
+ us2, err := db.UpsertUserSchedule(ctx, us1.ID, userID, scheduleID, &time2, nil, nil, json.RawMessage(`{"v":2}`))
require.NoError(t, err)
- // Same user + schedule = same user_schedule row
require.Equal(t, us1.ID, us2.ID)
require.WithinDuration(t, time2, *us2.ScheduledAt, time.Second)
}
+func TestUpsertUserScheduleCreatesDistinctInstances(t *testing.T) {
+ t.Parallel()
+
+ db := NewContainerStore(t)
+ projectID := uuid.New()
+ ctx := context.Background()
+
+ userID := createTestUserForSchedules(t, db, ctx, projectID)
+ scheduleID, err := db.UpsertSchedule(ctx, projectID, "multi_instance", "single")
+ require.NoError(t, err)
+
+ // Two upserts without an id create two distinct assignments for the same
+ // (user, schedule) — the same name can now be scheduled multiple times.
+ time1 := time.Now().Add(24 * time.Hour).UTC().Truncate(time.Microsecond)
+ us1, err := db.UpsertUserSchedule(ctx, uuid.Nil, userID, scheduleID, &time1, nil, nil, json.RawMessage(`{"v":1}`))
+ require.NoError(t, err)
+
+ time2 := time.Now().Add(48 * time.Hour).UTC().Truncate(time.Microsecond)
+ us2, err := db.UpsertUserSchedule(ctx, uuid.Nil, userID, scheduleID, &time2, nil, nil, json.RawMessage(`{"v":2}`))
+ require.NoError(t, err)
+
+ require.NotEqual(t, us1.ID, us2.ID)
+
+ all, err := db.ListUserSchedulesByScheduleID(ctx, userID, scheduleID)
+ require.NoError(t, err)
+ require.Len(t, all, 2)
+}
+
+func TestUpsertUserScheduleOwnershipMismatch(t *testing.T) {
+ t.Parallel()
+
+ db := NewContainerStore(t)
+ projectID := uuid.New()
+ ctx := context.Background()
+
+ userA := createTestUserForSchedules(t, db, ctx, projectID)
+ userB := createTestUserForSchedules(t, db, ctx, projectID)
+ scheduleID, err := db.UpsertSchedule(ctx, projectID, "ownership", "single")
+ require.NoError(t, err)
+
+ at := time.Now().Add(24 * time.Hour).UTC().Truncate(time.Microsecond)
+ owned, err := db.UpsertUserSchedule(ctx, uuid.Nil, userA, scheduleID, &at, nil, nil, json.RawMessage(`{}`))
+ require.NoError(t, err)
+
+ // Re-using userA's assignment id under a different user must be rejected
+ // rather than silently rewriting userA's row.
+ _, err = db.UpsertUserSchedule(ctx, owned.ID, userB, scheduleID, &at, nil, nil, json.RawMessage(`{}`))
+ require.ErrorIs(t, err, ErrScheduleOwnershipMismatch)
+}
+
func TestListUserSchedules(t *testing.T) {
t.Parallel()
@@ -646,20 +696,33 @@ func TestListUserSchedulesHasPendingEvents(t *testing.T) {
_, err = db.CreateUserSchedule(ctx, userID, scheduleID, &futureTime, nil, nil, json.RawMessage(`{}`))
require.NoError(t, err)
- // While the generated event is unfired, the schedule reports pending events.
+ // An unfired event whose fire_at is still in the future is NOT "sending":
+ // has_pending_events only reflects events that are actually due. (Regression:
+ // recurring schedules pre-generate the next occurrence's event, which kept the
+ // "Sending" badge lit indefinitely.)
items, _, err := db.ListUserSchedules(ctx, projectID, userID, store.Pagination{Limit: 10, Offset: 0})
require.NoError(t, err)
require.Len(t, items, 1)
- require.True(t, items[0].HasPendingEvents)
+ require.False(t, items[0].HasPendingEvents)
- // Firing every event must clear the flag (regression: the badge previously
- // relied on scheduled_at <= now and stayed lit forever).
events, err := db.ListPendingScheduledEventsForUser(ctx, userID, scheduleID)
require.NoError(t, err)
- require.NotEmpty(t, events)
- for _, e := range events {
- require.NoError(t, db.MarkScheduledEventFired(ctx, e.ID))
- }
+ require.Len(t, events, 1)
+
+ // Once the event is due (fire_at <= now), the schedule reports pending events.
+ pastTime := time.Now().Add(-1 * time.Hour).UTC().Truncate(time.Microsecond)
+ _, err = db.ScheduledStore.db.ExecContext(ctx,
+ `UPDATE user_scheduled_events SET fire_at = $1 WHERE id = $2`,
+ pastTime, events[0].ID)
+ require.NoError(t, err)
+
+ items, _, err = db.ListUserSchedules(ctx, projectID, userID, store.Pagination{Limit: 10, Offset: 0})
+ require.NoError(t, err)
+ require.Len(t, items, 1)
+ require.True(t, items[0].HasPendingEvents)
+
+ // Firing the event clears the flag.
+ require.NoError(t, db.MarkScheduledEventFired(ctx, events[0].ID))
items, _, err = db.ListUserSchedules(ctx, projectID, userID, store.Pagination{Limit: 10, Offset: 0})
require.NoError(t, err)
@@ -1128,14 +1191,14 @@ func TestUpsertOrganizationSchedule(t *testing.T) {
require.NoError(t, err)
time1 := time.Now().Add(24 * time.Hour).UTC().Truncate(time.Microsecond)
- os1, err := db.UpsertOrganizationSchedule(ctx, orgID, scheduleID, &time1, nil, nil, json.RawMessage(`{"v":1}`))
+ os1, err := db.UpsertOrganizationSchedule(ctx, uuid.Nil, orgID, scheduleID, &time1, nil, nil, json.RawMessage(`{"v":1}`))
require.NoError(t, err)
+ // Re-upserting with the same assignment id updates that row in place.
time2 := time.Now().Add(48 * time.Hour).UTC().Truncate(time.Microsecond)
- os2, err := db.UpsertOrganizationSchedule(ctx, orgID, scheduleID, &time2, nil, nil, json.RawMessage(`{"v":2}`))
+ os2, err := db.UpsertOrganizationSchedule(ctx, os1.ID, orgID, scheduleID, &time2, nil, nil, json.RawMessage(`{"v":2}`))
require.NoError(t, err)
- // Same org + schedule = same row
require.Equal(t, os1.ID, os2.ID)
require.WithinDuration(t, time2, *os2.ScheduledAt, time.Second)
}
@@ -1185,19 +1248,33 @@ func TestListOrganizationSchedulesHasPendingEvents(t *testing.T) {
_, err = db.CreateOrganizationSchedule(ctx, orgID, scheduleID, &futureTime, nil, nil, json.RawMessage(`{}`))
require.NoError(t, err)
- // While the generated event is unfired, the schedule reports pending events.
+ // An unfired event whose fire_at is still in the future is NOT "sending":
+ // has_pending_events only reflects events that are actually due. (Regression:
+ // recurring schedules pre-generate the next occurrence's event, which kept the
+ // "Sending" badge lit indefinitely.)
items, _, err := db.ListOrganizationSchedules(ctx, projectID, orgID, store.Pagination{Limit: 10, Offset: 0})
require.NoError(t, err)
require.Len(t, items, 1)
- require.True(t, items[0].HasPendingEvents)
+ require.False(t, items[0].HasPendingEvents)
- // Firing every event must clear the flag.
events, err := db.ListPendingOrgScheduledEventsForOrg(ctx, orgID, scheduleID)
require.NoError(t, err)
- require.NotEmpty(t, events)
- for _, e := range events {
- require.NoError(t, db.MarkOrgScheduledEventFired(ctx, e.ID))
- }
+ require.Len(t, events, 1)
+
+ // Once the event is due (fire_at <= now), the schedule reports pending events.
+ pastTime := time.Now().Add(-1 * time.Hour).UTC().Truncate(time.Microsecond)
+ _, err = db.ScheduledStore.db.ExecContext(ctx,
+ `UPDATE organization_scheduled_events SET fire_at = $1 WHERE id = $2`,
+ pastTime, events[0].ID)
+ require.NoError(t, err)
+
+ items, _, err = db.ListOrganizationSchedules(ctx, projectID, orgID, store.Pagination{Limit: 10, Offset: 0})
+ require.NoError(t, err)
+ require.Len(t, items, 1)
+ require.True(t, items[0].HasPendingEvents)
+
+ // Firing the event clears the flag.
+ require.NoError(t, db.MarkOrgScheduledEventFired(ctx, events[0].ID))
items, _, err = db.ListOrganizationSchedules(ctx, projectID, orgID, store.Pagination{Limit: 10, Offset: 0})
require.NoError(t, err)
@@ -1718,14 +1795,14 @@ func TestUpsertUserScheduleRecurring(t *testing.T) {
startAt := time.Now().Add(-14 * 24 * time.Hour).UTC().Truncate(time.Microsecond)
interval := "7 days"
- us, err := db.UpsertUserSchedule(ctx, userID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{}`))
+ us, err := db.UpsertUserSchedule(ctx, uuid.Nil, userID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{}`))
require.NoError(t, err)
require.NotNil(t, us.ScheduledAt)
require.True(t, us.ScheduledAt.After(time.Now()))
require.Greater(t, us.Occurrence, 0)
- // Upsert again with different data – same row
- us2, err := db.UpsertUserSchedule(ctx, userID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{"updated":true}`))
+ // Upsert again with the same id and different data – same row
+ us2, err := db.UpsertUserSchedule(ctx, us.ID, userID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{"updated":true}`))
require.NoError(t, err)
require.Equal(t, us.ID, us2.ID)
}
@@ -1744,13 +1821,13 @@ func TestUpsertOrganizationScheduleRecurring(t *testing.T) {
startAt := time.Now().Add(-14 * 24 * time.Hour).UTC().Truncate(time.Microsecond)
interval := "7 days"
- os, err := db.UpsertOrganizationSchedule(ctx, orgID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{}`))
+ os, err := db.UpsertOrganizationSchedule(ctx, uuid.Nil, orgID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{}`))
require.NoError(t, err)
require.NotNil(t, os.ScheduledAt)
require.True(t, os.ScheduledAt.After(time.Now()))
require.Greater(t, os.Occurrence, 0)
- os2, err := db.UpsertOrganizationSchedule(ctx, orgID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{"updated":true}`))
+ os2, err := db.UpsertOrganizationSchedule(ctx, os.ID, orgID, scheduleID, nil, &startAt, &interval, json.RawMessage(`{"updated":true}`))
require.NoError(t, err)
require.Equal(t, os.ID, os2.ID)
}
@@ -1832,3 +1909,144 @@ func TestOrgScheduleDataPropagatedToEvents(t *testing.T) {
require.Len(t, events, 1)
require.JSONEq(t, `{"meeting":"standup"}`, string(events[0].Data))
}
+
+// nextOccurrence calls the next_schedule_occurrence SQL function with an explicit
+// as_of so results are deterministic, and returns the occurrence number, its
+// timestamp, and whether the defining properties (1)-(3) hold for that result.
+type occurrenceProps struct {
+ Occurrence int64 `db:"occurrence"`
+ NextAt time.Time `db:"next_at"`
+ P1Future bool `db:"p1_future"`
+ P2First bool `db:"p2_first"`
+ P3Forward bool `db:"p3_forward"`
+}
+
+func nextOccurrence(t *testing.T, db *State, anchor time.Time, interval string, cur int64, asOf time.Time) occurrenceProps {
+ t.Helper()
+ // Properties are evaluated with interval arithmetic in Postgres (the only place
+ // calendar math is exact). next_at > as_of (in the future); the previous
+ // occurrence is at or before as_of unless it is the first one (it is the
+ // earliest such occurrence); and the occurrence never moves backwards.
+ const q = `
+ WITH r AS (
+ SELECT next_at, occurrence
+ FROM next_schedule_occurrence($1::timestamptz, $2::interval, $3::bigint, $4::timestamptz)
+ )
+ SELECT
+ r.occurrence,
+ r.next_at,
+ (r.next_at > $4) AS p1_future,
+ (r.occurrence = $3 + 1 OR $1::timestamptz + (r.occurrence - 1) * $2::interval <= $4) AS p2_first,
+ (r.occurrence >= $3 + 1) AS p3_forward
+ FROM r`
+
+ var pr occurrenceProps
+ err := db.ScheduledStore.db.QueryRowxContext(context.Background(), q, anchor, interval, cur, asOf).StructScan(&pr)
+ require.NoError(t, err)
+ return pr
+}
+
+func TestNextScheduleOccurrenceExactCases(t *testing.T) {
+ t.Parallel()
+
+ db := NewContainerStore(t)
+
+ ts := func(s string) time.Time {
+ v, err := time.Parse(time.RFC3339, s)
+ require.NoError(t, err)
+ return v.UTC()
+ }
+
+ type testCase struct {
+ anchor string
+ interval string
+ cur int64
+ asOf string
+ wantOcc int64
+ wantNext string
+ }
+
+ cases := map[string]testCase{
+ // The stuck-"Sending" anniversary: yearly, first send is anchor + 1 year.
+ "yearly anniversary": {"2026-03-28T15:45:52Z", "1 year", 0, "2026-06-25T00:00:00Z", 1, "2027-03-28T15:45:52Z"},
+ // Leap-day anchor + yearly clamps to Feb 28 in common years (occurrence 3 = 2027).
+ "leap-day yearly clamps to Feb 28": {"2024-02-29T12:00:00Z", "1 year", 0, "2027-01-01T00:00:00Z", 3, "2027-02-28T12:00:00Z"},
+ // Leap-day anchor + yearly lands back on Feb 29 every 4th year (occurrence 4 = 2028).
+ "leap-day yearly lands on Feb 29": {"2024-02-29T12:00:00Z", "1 year", 0, "2027-06-01T00:00:00Z", 4, "2028-02-29T12:00:00Z"},
+ // as_of exactly on an occurrence boundary must pick the next one (strict >).
+ "daily on boundary picks next": {"2026-01-01T00:00:00Z", "1 day", 0, "2026-01-10T00:00:00Z", 10, "2026-01-11T00:00:00Z"},
+ // Month-end clamping: Jan 31 + 1 month = Feb 29 (leap), + 2 months = Mar 31.
+ "month-end clamp from Jan 31": {"2024-01-31T12:00:00Z", "1 month", 0, "2024-03-15T00:00:00Z", 2, "2024-03-31T12:00:00Z"},
+ // Advancing from a non-zero current occurrence only moves forward.
+ "advance from current occurrence": {"2026-01-01T00:00:00Z", "1 day", 5, "2026-01-03T00:00:00Z", 6, "2026-01-07T00:00:00Z"},
+ // ~470 years of dormancy on a 1-minute interval: far past the old 10000 cap.
+ "far dormancy beyond old cap": {"2026-01-01T00:00:00Z", "1 minute", 0, "2496-01-01T00:00:00Z", 247196161, "2496-01-01T00:01:00Z"},
+ }
+
+ for name, tc := range cases {
+ t.Run(name, func(t *testing.T) {
+ pr := nextOccurrence(t, db, ts(tc.anchor), tc.interval, tc.cur, ts(tc.asOf))
+ require.Equal(t, tc.wantOcc, pr.Occurrence)
+ require.Truef(t, pr.NextAt.Equal(ts(tc.wantNext)), "next_at = %s, want %s", pr.NextAt, tc.wantNext)
+ require.True(t, pr.P1Future && pr.P2First && pr.P3Forward, "defining properties must hold")
+ })
+ }
+}
+
+// TestNextScheduleOccurrenceProperties asserts the result is exact across a broad
+// grid (leap years, month-end clamping, mixed calendar intervals, horizons out to
+// year 3025) by checking the three defining properties for every combination.
+func TestNextScheduleOccurrenceProperties(t *testing.T) {
+ t.Parallel()
+
+ db := NewContainerStore(t)
+
+ ts := func(s string) time.Time {
+ v, err := time.Parse(time.RFC3339, s)
+ require.NoError(t, err)
+ return v.UTC()
+ }
+
+ anchors := []string{
+ "2024-01-31T12:00:00Z", "2020-12-31T06:00:00Z",
+ "2026-03-28T15:45:52Z", "1999-11-15T09:00:00Z",
+ }
+ intervals := []string{
+ "1 minute", "1 hour", "1 day", "1 week", "30 days",
+ "1 month", "3 months", "1 year", "5 years", "1 mon 15 days",
+ }
+ curs := []int64{0, 1, 7, 99}
+ asOfs := []string{
+ "2026-06-25T14:00:00Z", "2030-02-28T00:00:00Z",
+ "2520-06-01T00:00:00Z", "3025-09-09T00:00:00Z",
+ }
+
+ for _, a := range anchors {
+ for _, iv := range intervals {
+ for _, cur := range curs {
+ for _, asOf := range asOfs {
+ pr := nextOccurrence(t, db, ts(a), iv, cur, ts(asOf))
+ require.Truef(t, pr.P1Future && pr.P2First && pr.P3Forward,
+ "anchor=%s interval=%s cur=%d asOf=%s -> occ=%d (p1=%v p2=%v p3=%v)",
+ a, iv, cur, asOf, pr.Occurrence, pr.P1Future, pr.P2First, pr.P3Forward)
+ }
+ }
+ }
+ }
+}
+
+func TestNextScheduleOccurrenceRejectsNonPositiveInterval(t *testing.T) {
+ t.Parallel()
+
+ db := NewContainerStore(t)
+
+ const q = `SELECT next_at FROM next_schedule_occurrence($1::timestamptz, $2::interval, 0, $3::timestamptz)`
+ for _, interval := range []string{"0 seconds", "-1 day"} {
+ var nextAt time.Time
+ err := db.ScheduledStore.db.QueryRowxContext(context.Background(), q,
+ time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), interval,
+ time.Date(2026, 6, 25, 0, 0, 0, 0, time.UTC)).Scan(&nextAt)
+ require.Error(t, err, "interval %q should be rejected", interval)
+ require.Contains(t, err.Error(), "interval must be positive")
+ }
+}