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/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/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/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/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 9cd88aaf..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 } @@ -1082,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 @@ -1101,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 } diff --git a/internal/store/subjects/scheduled_test.go b/internal/store/subjects/scheduled_test.go index cbc0dc01..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() @@ -1141,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) } @@ -1745,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) } @@ -1771,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) }