package mongo import ( "context" "errors" "time" "go.mongodb.org/mongo-driver/bson" mongodriver "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" "go.mongodb.org/mongo-driver/bson/primitive" "scheduler-backend/internal/store/model" ) type JobConfigStore struct { collection *mongodriver.Collection } func NewJobConfigStore(db *mongodriver.Database) *JobConfigStore { return &JobConfigStore{ collection: db.Collection("job_configs"), } } func (s *JobConfigStore) List(ctx context.Context) ([]model.JobConfig, error) { findCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() cursor, err := s.collection.Find( findCtx, bson.D{}, options.Find().SetSort(bson.D{{Key: "updatedAt", Value: -1}}), ) if err != nil { return nil, err } defer cursor.Close(findCtx) items := make([]model.JobConfig, 0) for cursor.Next(findCtx) { var item model.JobConfig if err := cursor.Decode(&item); err != nil { return nil, err } items = append(items, item) } if err := cursor.Err(); err != nil { return nil, err } return items, nil } func (s *JobConfigStore) ListEnabled(ctx context.Context) ([]model.JobConfig, error) { findCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() cursor, err := s.collection.Find( findCtx, bson.M{"enabled": true}, options.Find().SetSort(bson.D{{Key: "updatedAt", Value: -1}}), ) if err != nil { return nil, err } defer cursor.Close(findCtx) items := make([]model.JobConfig, 0) for cursor.Next(findCtx) { var item model.JobConfig if err := cursor.Decode(&item); err != nil { return nil, err } items = append(items, item) } if err := cursor.Err(); err != nil { return nil, err } return items, nil } func (s *JobConfigStore) GetByID(ctx context.Context, id primitive.ObjectID) (*model.JobConfig, error) { findCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() var item model.JobConfig err := s.collection.FindOne(findCtx, bson.M{"_id": id}).Decode(&item) if err != nil { if errors.Is(err, mongodriver.ErrNoDocuments) { return nil, nil } return nil, err } return &item, nil } func (s *JobConfigStore) Create(ctx context.Context, item *model.JobConfig) error { insertCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() now := time.Now() item.CreatedAt = now item.UpdatedAt = now if item.LastStatus == "" { item.LastStatus = "idle" } result, err := s.collection.InsertOne(insertCtx, item) if err != nil { return err } if oid, ok := result.InsertedID.(primitive.ObjectID); ok { item.ID = oid } return nil } func (s *JobConfigStore) Update(ctx context.Context, item *model.JobConfig) error { updateCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() item.UpdatedAt = time.Now() update := bson.M{ "$set": bson.M{ "name": item.Name, "handlerKey": item.HandlerKey, "enabled": item.Enabled, "scheduleType": item.ScheduleType, "scheduleValue": item.ScheduleValue, "defaultParams": item.DefaultParams, "updatedAt": item.UpdatedAt, }, } _, err := s.collection.UpdateByID(updateCtx, item.ID, update) return err } func (s *JobConfigStore) ToggleEnabled(ctx context.Context, id primitive.ObjectID, enabled bool) error { updateCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() _, err := s.collection.UpdateByID(updateCtx, id, bson.M{ "$set": bson.M{ "enabled": enabled, "updatedAt": time.Now(), }, }) return err } func (s *JobConfigStore) UpdateRunState(ctx context.Context, id primitive.ObjectID, status string, lastRunAt *time.Time) error { updateCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() set := bson.M{ "lastStatus": status, "updatedAt": time.Now(), } if lastRunAt != nil { set["lastRunAt"] = *lastRunAt } _, err := s.collection.UpdateByID(updateCtx, id, bson.M{ "$set": set, }) return err } func (s *JobConfigStore) UpsertByHandlerKey(ctx context.Context, item *model.JobConfig) error { upsertCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() now := time.Now() _, err := s.collection.UpdateOne( upsertCtx, bson.M{"handlerKey": item.HandlerKey}, bson.M{ "$set": bson.M{ "name": item.Name, "handlerKey": item.HandlerKey, "scheduleType": item.ScheduleType, "scheduleValue": item.ScheduleValue, "defaultParams": item.DefaultParams, "updatedAt": now, }, "$setOnInsert": bson.M{ "enabled": false, "lastStatus": "idle", "createdAt": now, }, }, options.Update().SetUpsert(true), ) return err } func (s *JobConfigStore) UpdateNextRunAt(ctx context.Context, id primitive.ObjectID, nextRunAt *time.Time) error { updateCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() set := bson.M{ "updatedAt": time.Now(), } if nextRunAt != nil { set["nextRunAt"] = *nextRunAt } else { set["nextRunAt"] = nil } _, err := s.collection.UpdateByID(updateCtx, id, bson.M{ "$set": set, }) return err }