package storage import ( "context" "fmt" "net/http" "strings" "github.com/atlasos/calypso/internal/common/cache" "github.com/atlasos/calypso/internal/common/database" "github.com/atlasos/calypso/internal/common/logger" "github.com/atlasos/calypso/internal/tasks" "github.com/gin-gonic/gin" ) // Handler handles storage-related API requests type Handler struct { diskService *DiskService lvmService *LVMService zfsService *ZFSService snapshotService *SnapshotService snapshotScheduleService *SnapshotScheduleService replicationService *ReplicationService arcService *ARCService taskEngine *tasks.Engine db *database.DB logger *logger.Logger cache *cache.Cache // Cache for invalidation } // SetCache sets the cache instance for cache invalidation func (h *Handler) SetCache(c *cache.Cache) { h.cache = c } // NewHandler creates a new storage handler func NewHandler(db *database.DB, log *logger.Logger) *Handler { snapshotService := NewSnapshotService(db, log) return &Handler{ diskService: NewDiskService(db, log), lvmService: NewLVMService(db, log), zfsService: NewZFSService(db, log), snapshotService: snapshotService, snapshotScheduleService: NewSnapshotScheduleService(db, log, snapshotService), replicationService: NewReplicationService(db, log), arcService: NewARCService(log), taskEngine: tasks.NewEngine(db, log), db: db, logger: log, } } // ListDisks lists all physical disks from database func (h *Handler) ListDisks(c *gin.Context) { disks, err := h.diskService.ListDisksFromDatabase(c.Request.Context()) if err != nil { h.logger.Error("Failed to list disks", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list disks"}) return } c.JSON(http.StatusOK, gin.H{"disks": disks}) } // SyncDisks syncs discovered disks to database func (h *Handler) SyncDisks(c *gin.Context) { userID, _ := c.Get("user_id") // Create async task taskID, err := h.taskEngine.CreateTask(c.Request.Context(), tasks.TaskTypeRescan, userID.(string), map[string]interface{}{ "operation": "sync_disks", }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create task"}) return } // Run sync in background go func() { // Create new context for background task (don't use request context which may expire) ctx := context.Background() h.taskEngine.StartTask(ctx, taskID) h.taskEngine.UpdateProgress(ctx, taskID, 50, "Discovering disks...") h.logger.Info("Starting disk sync", "task_id", taskID) if err := h.diskService.SyncDisksToDatabase(ctx); err != nil { h.logger.Error("Disk sync failed", "task_id", taskID, "error", err) h.taskEngine.FailTask(ctx, taskID, err.Error()) return } h.logger.Info("Disk sync completed", "task_id", taskID) h.taskEngine.UpdateProgress(ctx, taskID, 100, "Disk sync completed") h.taskEngine.CompleteTask(ctx, taskID, "Disks synchronized successfully") }() c.JSON(http.StatusAccepted, gin.H{"task_id": taskID}) } // ListVolumeGroups lists all volume groups func (h *Handler) ListVolumeGroups(c *gin.Context) { vgs, err := h.lvmService.ListVolumeGroups(c.Request.Context()) if err != nil { h.logger.Error("Failed to list volume groups", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list volume groups"}) return } c.JSON(http.StatusOK, gin.H{"volume_groups": vgs}) } // ListRepositories lists all repositories func (h *Handler) ListRepositories(c *gin.Context) { repos, err := h.lvmService.ListRepositories(c.Request.Context()) if err != nil { h.logger.Error("Failed to list repositories", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list repositories"}) return } c.JSON(http.StatusOK, gin.H{"repositories": repos}) } // GetRepository retrieves a repository by ID func (h *Handler) GetRepository(c *gin.Context) { repoID := c.Param("id") repo, err := h.lvmService.GetRepository(c.Request.Context(), repoID) if err != nil { if err.Error() == "repository not found" { c.JSON(http.StatusNotFound, gin.H{"error": "repository not found"}) return } h.logger.Error("Failed to get repository", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get repository"}) return } c.JSON(http.StatusOK, repo) } // CreateRepositoryRequest represents a repository creation request type CreateRepositoryRequest struct { Name string `json:"name" binding:"required"` Description string `json:"description"` VolumeGroup string `json:"volume_group" binding:"required"` SizeGB int64 `json:"size_gb" binding:"required"` } // CreateRepository creates a new repository func (h *Handler) CreateRepository(c *gin.Context) { var req CreateRepositoryRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request"}) return } userID, _ := c.Get("user_id") sizeBytes := req.SizeGB * 1024 * 1024 * 1024 repo, err := h.lvmService.CreateRepository( c.Request.Context(), req.Name, req.VolumeGroup, sizeBytes, userID.(string), ) if err != nil { h.logger.Error("Failed to create repository", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusCreated, repo) } // DeleteRepository deletes a repository func (h *Handler) DeleteRepository(c *gin.Context) { repoID := c.Param("id") if err := h.lvmService.DeleteRepository(c.Request.Context(), repoID); err != nil { if err.Error() == "repository not found" { c.JSON(http.StatusNotFound, gin.H{"error": "repository not found"}) return } h.logger.Error("Failed to delete repository", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "repository deleted successfully"}) } // CreateZPoolRequest represents a ZFS pool creation request type CreateZPoolRequest struct { Name string `json:"name" binding:"required"` Description string `json:"description"` RaidLevel string `json:"raid_level" binding:"required"` // stripe, mirror, raidz, raidz2, raidz3 Disks []string `json:"disks" binding:"required"` // device paths Compression string `json:"compression"` // off, lz4, zstd, gzip Deduplication bool `json:"deduplication"` AutoExpand bool `json:"auto_expand"` } // CreateZPool creates a new ZFS pool func (h *Handler) CreateZPool(c *gin.Context) { var req CreateZPoolRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid request body", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } // Validate required fields if req.Name == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "pool name is required"}) return } if req.RaidLevel == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "raid_level is required"}) return } if len(req.Disks) == 0 { c.JSON(http.StatusBadRequest, gin.H{"error": "at least one disk is required"}) return } userID, exists := c.Get("user_id") if !exists { h.logger.Error("User ID not found in context") c.JSON(http.StatusUnauthorized, gin.H{"error": "authentication required"}) return } userIDStr, ok := userID.(string) if !ok { h.logger.Error("Invalid user ID type", "type", fmt.Sprintf("%T", userID)) c.JSON(http.StatusInternalServerError, gin.H{"error": "invalid user context"}) return } // Set default compression if not provided if req.Compression == "" { req.Compression = "lz4" } h.logger.Info("Creating ZFS pool request", "name", req.Name, "raid_level", req.RaidLevel, "disks", req.Disks, "compression", req.Compression) pool, err := h.zfsService.CreatePool( c.Request.Context(), req.Name, req.RaidLevel, req.Disks, req.Compression, req.Deduplication, req.AutoExpand, userIDStr, ) if err != nil { h.logger.Error("Failed to create ZFS pool", "error", err, "name", req.Name, "raid_level", req.RaidLevel, "disks", req.Disks) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } h.logger.Info("ZFS pool created successfully", "pool_id", pool.ID, "name", pool.Name) c.JSON(http.StatusCreated, pool) } // ListZFSPools lists all ZFS pools func (h *Handler) ListZFSPools(c *gin.Context) { pools, err := h.zfsService.ListPools(c.Request.Context()) if err != nil { h.logger.Error("Failed to list ZFS pools", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list ZFS pools"}) return } c.JSON(http.StatusOK, gin.H{"pools": pools}) } // GetZFSPool retrieves a ZFS pool by ID func (h *Handler) GetZFSPool(c *gin.Context) { poolID := c.Param("id") pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get ZFS pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get ZFS pool"}) return } c.JSON(http.StatusOK, pool) } // DeleteZFSPool deletes a ZFS pool func (h *Handler) DeleteZFSPool(c *gin.Context) { poolID := c.Param("id") if err := h.zfsService.DeletePool(c.Request.Context(), poolID); err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to delete ZFS pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Invalidate cache for pools list if h.cache != nil { cacheKey := "http:/api/v1/storage/zfs/pools:" h.cache.Delete(cacheKey) h.logger.Debug("Cache invalidated for pools list", "key", cacheKey) } c.JSON(http.StatusOK, gin.H{"message": "ZFS pool deleted successfully"}) } // AddSpareDiskRequest represents a request to add spare disks to a pool type AddSpareDiskRequest struct { Disks []string `json:"disks" binding:"required"` } // AddSpareDisk adds spare disks to a ZFS pool func (h *Handler) AddSpareDisk(c *gin.Context) { poolID := c.Param("id") var req AddSpareDiskRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid add spare disk request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } if len(req.Disks) == 0 { c.JSON(http.StatusBadRequest, gin.H{"error": "at least one disk must be specified"}) return } if err := h.zfsService.AddSpareDisk(c.Request.Context(), poolID, req.Disks); err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to add spare disks", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "Spare disks added successfully"}) } // ListZFSDatasets lists all datasets in a ZFS pool func (h *Handler) ListZFSDatasets(c *gin.Context) { poolID := c.Param("id") // Get pool to get pool name pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get pool"}) return } datasets, err := h.zfsService.ListDatasets(c.Request.Context(), pool.Name) if err != nil { h.logger.Error("Failed to list datasets", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Ensure we return an empty array instead of null if datasets == nil { datasets = []*ZFSDataset{} } c.JSON(http.StatusOK, gin.H{"datasets": datasets}) } // CreateZFSDatasetRequest represents a request to create a ZFS dataset type CreateZFSDatasetRequest struct { Name string `json:"name" binding:"required"` // Dataset name (without pool prefix) Type string `json:"type" binding:"required"` // "filesystem" or "volume" Compression string `json:"compression"` // off, lz4, zstd, gzip, etc. Quota int64 `json:"quota"` // -1 for unlimited, >0 for size Reservation int64 `json:"reservation"` // 0 for none MountPoint string `json:"mount_point"` // Optional mount point } // CreateZFSDataset creates a new ZFS dataset in a pool func (h *Handler) CreateZFSDataset(c *gin.Context) { poolID := c.Param("id") // Get pool to get pool name pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get pool"}) return } var req CreateZFSDatasetRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid create dataset request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } // Validate type if req.Type != "filesystem" && req.Type != "volume" { c.JSON(http.StatusBadRequest, gin.H{"error": "type must be 'filesystem' or 'volume'"}) return } // Validate mount point: volumes cannot have mount points if req.Type == "volume" && req.MountPoint != "" { c.JSON(http.StatusBadRequest, gin.H{"error": "mount point cannot be set for volume datasets (volumes are block devices for iSCSI export)"}) return } // Validate dataset name (should not contain pool name) if strings.Contains(req.Name, "/") { c.JSON(http.StatusBadRequest, gin.H{"error": "dataset name should not contain '/' (pool name is automatically prepended)"}) return } // Create dataset request - CreateDatasetRequest is in the same package (zfs.go) createReq := CreateDatasetRequest{ Name: req.Name, Type: req.Type, Compression: req.Compression, Quota: req.Quota, Reservation: req.Reservation, MountPoint: req.MountPoint, } dataset, err := h.zfsService.CreateDataset(c.Request.Context(), pool.Name, createReq) if err != nil { h.logger.Error("Failed to create dataset", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusCreated, dataset) } // DeleteZFSDataset deletes a ZFS dataset func (h *Handler) DeleteZFSDataset(c *gin.Context) { poolID := c.Param("id") datasetName := c.Param("dataset") // Get pool to get pool name pool, err := h.zfsService.GetPool(c.Request.Context(), poolID) if err != nil { if err.Error() == "pool not found" { c.JSON(http.StatusNotFound, gin.H{"error": "pool not found"}) return } h.logger.Error("Failed to get pool", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get pool"}) return } // Construct full dataset name fullDatasetName := pool.Name + "/" + datasetName // Verify dataset belongs to this pool if !strings.HasPrefix(fullDatasetName, pool.Name+"/") { c.JSON(http.StatusBadRequest, gin.H{"error": "dataset does not belong to this pool"}) return } if err := h.zfsService.DeleteDataset(c.Request.Context(), fullDatasetName); err != nil { if strings.Contains(err.Error(), "does not exist") || strings.Contains(err.Error(), "not found") { c.JSON(http.StatusNotFound, gin.H{"error": "dataset not found"}) return } h.logger.Error("Failed to delete dataset", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Invalidate cache for this pool's datasets list if h.cache != nil { // Generate cache key using the same format as cache middleware cacheKey := fmt.Sprintf("http:/api/v1/storage/zfs/pools/%s/datasets:", poolID) h.cache.Delete(cacheKey) // Also invalidate any cached responses with query parameters h.logger.Debug("Cache invalidated for dataset list", "pool_id", poolID, "key", cacheKey) } c.JSON(http.StatusOK, gin.H{"message": "Dataset deleted successfully"}) } // GetARCStats returns ZFS ARC statistics func (h *Handler) GetARCStats(c *gin.Context) { stats, err := h.arcService.GetARCStats(c.Request.Context()) if err != nil { h.logger.Error("Failed to get ARC stats", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get ARC stats: " + err.Error()}) return } c.JSON(http.StatusOK, stats) } // ListSnapshots lists all snapshots, optionally filtered by dataset func (h *Handler) ListSnapshots(c *gin.Context) { datasetFilter := c.DefaultQuery("dataset", "") snapshots, err := h.snapshotService.ListSnapshots(c.Request.Context(), datasetFilter) if err != nil { h.logger.Error("Failed to list snapshots", "error", err, "dataset", datasetFilter) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list snapshots: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"snapshots": snapshots}) } // CreateSnapshotRequest represents a request to create a snapshot type CreateSnapshotRequest struct { Dataset string `json:"dataset" binding:"required"` Name string `json:"name" binding:"required"` Recursive bool `json:"recursive"` } // CreateSnapshot creates a new snapshot func (h *Handler) CreateSnapshot(c *gin.Context) { var req CreateSnapshotRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid create snapshot request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } if err := h.snapshotService.CreateSnapshot(c.Request.Context(), req.Dataset, req.Name, req.Recursive); err != nil { h.logger.Error("Failed to create snapshot", "error", err, "dataset", req.Dataset, "name", req.Name) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create snapshot: " + err.Error()}) return } c.JSON(http.StatusCreated, gin.H{"message": "snapshot created successfully"}) } // DeleteSnapshot deletes a snapshot func (h *Handler) DeleteSnapshot(c *gin.Context) { snapshotName := c.Param("name") recursive := c.DefaultQuery("recursive", "false") == "true" if err := h.snapshotService.DeleteSnapshot(c.Request.Context(), snapshotName, recursive); err != nil { h.logger.Error("Failed to delete snapshot", "error", err, "snapshot", snapshotName) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete snapshot: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "snapshot deleted successfully"}) } // RollbackSnapshotRequest represents a request to rollback to a snapshot type RollbackSnapshotRequest struct { Force bool `json:"force"` } // RollbackSnapshot rolls back a dataset to a snapshot func (h *Handler) RollbackSnapshot(c *gin.Context) { snapshotName := c.Param("name") var req RollbackSnapshotRequest if err := c.ShouldBindJSON(&req); err != nil { // Default to false if not provided req.Force = false } if err := h.snapshotService.RollbackSnapshot(c.Request.Context(), snapshotName, req.Force); err != nil { h.logger.Error("Failed to rollback snapshot", "error", err, "snapshot", snapshotName) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to rollback snapshot: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "snapshot rollback completed successfully"}) } // CloneSnapshotRequest represents a request to clone a snapshot type CloneSnapshotRequest struct { CloneName string `json:"clone_name" binding:"required"` } // CloneSnapshot clones a snapshot to a new dataset func (h *Handler) CloneSnapshot(c *gin.Context) { snapshotName := c.Param("name") var req CloneSnapshotRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid clone snapshot request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } if err := h.snapshotService.CloneSnapshot(c.Request.Context(), snapshotName, req.CloneName); err != nil { h.logger.Error("Failed to clone snapshot", "error", err, "snapshot", snapshotName, "clone", req.CloneName) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to clone snapshot: " + err.Error()}) return } c.JSON(http.StatusCreated, gin.H{"message": "snapshot cloned successfully", "clone_name": req.CloneName}) } // ListSnapshotSchedules lists all snapshot schedules func (h *Handler) ListSnapshotSchedules(c *gin.Context) { schedules, err := h.snapshotScheduleService.ListSchedules(c.Request.Context()) if err != nil { h.logger.Error("Failed to list snapshot schedules", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list snapshot schedules: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"schedules": schedules}) } // GetSnapshotSchedule retrieves a snapshot schedule by ID func (h *Handler) GetSnapshotSchedule(c *gin.Context) { id := c.Param("id") schedule, err := h.snapshotScheduleService.GetSchedule(c.Request.Context(), id) if err != nil { if err.Error() == "schedule not found" { c.JSON(http.StatusNotFound, gin.H{"error": "schedule not found"}) return } h.logger.Error("Failed to get snapshot schedule", "error", err, "id", id) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get snapshot schedule: " + err.Error()}) return } c.JSON(http.StatusOK, schedule) } // CreateSnapshotScheduleRequest represents a request to create a snapshot schedule type CreateSnapshotScheduleRequest struct { Name string `json:"name" binding:"required"` Dataset string `json:"dataset" binding:"required"` SnapshotNameTemplate string `json:"snapshot_name_template" binding:"required"` ScheduleType string `json:"schedule_type" binding:"required"` ScheduleConfig map[string]interface{} `json:"schedule_config" binding:"required"` Recursive bool `json:"recursive"` RetentionCount *int `json:"retention_count"` RetentionDays *int `json:"retention_days"` } // CreateSnapshotSchedule creates a new snapshot schedule func (h *Handler) CreateSnapshotSchedule(c *gin.Context) { var req CreateSnapshotScheduleRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid create snapshot schedule request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } userID, _ := c.Get("user_id") userIDStr := "" if userID != nil { userIDStr = userID.(string) } schedule, err := h.snapshotScheduleService.CreateSchedule(c.Request.Context(), &CreateScheduleRequest{ Name: req.Name, Dataset: req.Dataset, SnapshotNameTemplate: req.SnapshotNameTemplate, ScheduleType: req.ScheduleType, ScheduleConfig: req.ScheduleConfig, Recursive: req.Recursive, RetentionCount: req.RetentionCount, RetentionDays: req.RetentionDays, }, userIDStr) if err != nil { h.logger.Error("Failed to create snapshot schedule", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create snapshot schedule: " + err.Error()}) return } c.JSON(http.StatusCreated, schedule) } // UpdateSnapshotSchedule updates an existing snapshot schedule func (h *Handler) UpdateSnapshotSchedule(c *gin.Context) { id := c.Param("id") var req CreateSnapshotScheduleRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid update snapshot schedule request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } schedule, err := h.snapshotScheduleService.UpdateSchedule(c.Request.Context(), id, &CreateScheduleRequest{ Name: req.Name, Dataset: req.Dataset, SnapshotNameTemplate: req.SnapshotNameTemplate, ScheduleType: req.ScheduleType, ScheduleConfig: req.ScheduleConfig, Recursive: req.Recursive, RetentionCount: req.RetentionCount, RetentionDays: req.RetentionDays, }) if err != nil { if err.Error() == "schedule not found" { c.JSON(http.StatusNotFound, gin.H{"error": "schedule not found"}) return } h.logger.Error("Failed to update snapshot schedule", "error", err, "id", id) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update snapshot schedule: " + err.Error()}) return } c.JSON(http.StatusOK, schedule) } // DeleteSnapshotSchedule deletes a snapshot schedule func (h *Handler) DeleteSnapshotSchedule(c *gin.Context) { id := c.Param("id") if err := h.snapshotScheduleService.DeleteSchedule(c.Request.Context(), id); err != nil { if err.Error() == "schedule not found" { c.JSON(http.StatusNotFound, gin.H{"error": "schedule not found"}) return } h.logger.Error("Failed to delete snapshot schedule", "error", err, "id", id) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete snapshot schedule: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "snapshot schedule deleted successfully"}) } // ToggleSnapshotSchedule enables or disables a snapshot schedule func (h *Handler) ToggleSnapshotSchedule(c *gin.Context) { id := c.Param("id") var req struct { Enabled bool `json:"enabled" binding:"required"` } if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid toggle snapshot schedule request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } if err := h.snapshotScheduleService.ToggleSchedule(c.Request.Context(), id, req.Enabled); err != nil { if err.Error() == "schedule not found" { c.JSON(http.StatusNotFound, gin.H{"error": "schedule not found"}) return } h.logger.Error("Failed to toggle snapshot schedule", "error", err, "id", id) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to toggle snapshot schedule: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "snapshot schedule toggled successfully"}) } // ListReplicationTasks lists all replication tasks func (h *Handler) ListReplicationTasks(c *gin.Context) { direction := c.Query("direction") // Optional filter: "outbound" or "inbound" tasks, err := h.replicationService.ListReplicationTasks(c.Request.Context(), direction) if err != nil { h.logger.Error("Failed to list replication tasks", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list replication tasks: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"tasks": tasks}) } // GetReplicationTask retrieves a replication task by ID func (h *Handler) GetReplicationTask(c *gin.Context) { id := c.Param("id") task, err := h.replicationService.GetReplicationTask(c.Request.Context(), id) if err != nil { if err.Error() == "replication task not found" { c.JSON(http.StatusNotFound, gin.H{"error": "replication task not found"}) return } h.logger.Error("Failed to get replication task", "error", err, "id", id) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get replication task: " + err.Error()}) return } c.JSON(http.StatusOK, task) } // CreateReplicationTaskRequest represents a request to create a replication task type CreateReplicationTaskRequest struct { Name string `json:"name" binding:"required"` Direction string `json:"direction" binding:"required"` SourceDataset *string `json:"source_dataset"` TargetHost *string `json:"target_host"` TargetPort *int `json:"target_port"` TargetUser *string `json:"target_user"` TargetDataset *string `json:"target_dataset"` TargetSSHKeyPath *string `json:"target_ssh_key_path"` SourceHost *string `json:"source_host"` SourcePort *int `json:"source_port"` SourceUser *string `json:"source_user"` LocalDataset *string `json:"local_dataset"` ScheduleType *string `json:"schedule_type"` ScheduleConfig map[string]interface{} `json:"schedule_config"` Compression string `json:"compression"` Encryption bool `json:"encryption"` Recursive bool `json:"recursive"` Incremental bool `json:"incremental"` AutoSnapshot bool `json:"auto_snapshot"` Enabled bool `json:"enabled"` } // CreateReplicationTask creates a new replication task func (h *Handler) CreateReplicationTask(c *gin.Context) { var req CreateReplicationTaskRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid create replication task request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } userID, _ := c.Get("user_id") userIDStr := "" if userID != nil { userIDStr = userID.(string) } task, err := h.replicationService.CreateReplicationTask(c.Request.Context(), &CreateReplicationRequest{ Name: req.Name, Direction: req.Direction, SourceDataset: req.SourceDataset, TargetHost: req.TargetHost, TargetPort: req.TargetPort, TargetUser: req.TargetUser, TargetDataset: req.TargetDataset, TargetSSHKeyPath: req.TargetSSHKeyPath, SourceHost: req.SourceHost, SourcePort: req.SourcePort, SourceUser: req.SourceUser, LocalDataset: req.LocalDataset, ScheduleType: req.ScheduleType, ScheduleConfig: req.ScheduleConfig, Compression: req.Compression, Encryption: req.Encryption, Recursive: req.Recursive, Incremental: req.Incremental, AutoSnapshot: req.AutoSnapshot, Enabled: req.Enabled, }, userIDStr) if err != nil { h.logger.Error("Failed to create replication task", "error", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create replication task: " + err.Error()}) return } c.JSON(http.StatusCreated, task) } // UpdateReplicationTask updates an existing replication task func (h *Handler) UpdateReplicationTask(c *gin.Context) { id := c.Param("id") var req CreateReplicationTaskRequest if err := c.ShouldBindJSON(&req); err != nil { h.logger.Error("Invalid update replication task request", "error", err) c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } task, err := h.replicationService.UpdateReplicationTask(c.Request.Context(), id, &CreateReplicationRequest{ Name: req.Name, Direction: req.Direction, SourceDataset: req.SourceDataset, TargetHost: req.TargetHost, TargetPort: req.TargetPort, TargetUser: req.TargetUser, TargetDataset: req.TargetDataset, TargetSSHKeyPath: req.TargetSSHKeyPath, SourceHost: req.SourceHost, SourcePort: req.SourcePort, SourceUser: req.SourceUser, LocalDataset: req.LocalDataset, ScheduleType: req.ScheduleType, ScheduleConfig: req.ScheduleConfig, Compression: req.Compression, Encryption: req.Encryption, Recursive: req.Recursive, Incremental: req.Incremental, AutoSnapshot: req.AutoSnapshot, Enabled: req.Enabled, }) if err != nil { if err.Error() == "replication task not found" { c.JSON(http.StatusNotFound, gin.H{"error": "replication task not found"}) return } h.logger.Error("Failed to update replication task", "error", err, "id", id) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update replication task: " + err.Error()}) return } c.JSON(http.StatusOK, task) } // DeleteReplicationTask deletes a replication task func (h *Handler) DeleteReplicationTask(c *gin.Context) { id := c.Param("id") if err := h.replicationService.DeleteReplicationTask(c.Request.Context(), id); err != nil { if err.Error() == "replication task not found" { c.JSON(http.StatusNotFound, gin.H{"error": "replication task not found"}) return } h.logger.Error("Failed to delete replication task", "error", err, "id", id) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete replication task: " + err.Error()}) return } c.JSON(http.StatusOK, gin.H{"message": "replication task deleted successfully"}) }