@@ -195,21 +195,21 @@ templ Dashboard(ctx context.Context, data DashboardData) {
- for _, service := range data.NotificationServices {
-
-
-
- if service.Type == "email" {
-
-
+
+
+ for _, service := range data.NotificationServices {
+
+
+
+
+
+ if service.Type == "email" {
+
+
+ } else if service.Type == "webhook" {
+
+
+ } else {
+
+
+ }
+
+
+ { service.Name }
+
+
+ { service.Description }
+
+
+
+
+
+
+
+
+
- } else if service.Type == "slack" {
-
-
+
+
+
+
+ if service.IsEnabled {
+ Active
+ } else {
+ Disabled
+ }
+
+
+ { service.Type }
+
+ if len(service.EventTriggers) > 0 && service.Type == "webhook" {
+
+ { fmt.Sprintf("%d triggers", len(service.EventTriggers)) }
+
+ }
+ if service.SuccessCount > 0 || service.FailureCount > 0 {
+
+ { fmt.Sprintf("%d/%d", service.SuccessCount, service.SuccessCount + service.FailureCount) }
+
+ }
+
+
+
+ if service.Type == "webhook" {
+
+
+ Events:
+
+ if len(service.EventTriggers) == 0 {
+ None
+ } else {
+ for i, trigger := range service.EventTriggers {
+ if i > 0 {
+ ,
+ }
+ { trigger }
+ }
+ }
+
+
+
+ Retry:
+
+ if service.RetryPolicy == "" {
+ Default
+ } else {
+ { service.RetryPolicy }
+ }
+
+
+
+ } else {
+
+
+
Last sent:
+ if service.SuccessCount > 0 {
+ "Recently"
+ } else {
+ "Never"
+ }
+
+
+ }
- } else if service.Type == "webhook" {
-
-
- } else {
-
-
- }
-
{ service.Name }
-
-
-
-
-
- if service.IsEnabled {
- Active
- } else {
- Disabled
- }
-
-
- { service.Type }
-
- if len(service.EventTriggers) > 0 && service.Type == "webhook" {
-
- { fmt.Sprintf("%d triggers", len(service.EventTriggers)) }
-
- }
- if service.SuccessCount > 0 || service.FailureCount > 0 {
-
- { fmt.Sprintf("%d/%d", service.SuccessCount, service.SuccessCount + service.FailureCount) }
-
- }
-
-
{ service.Description }
- if service.Type == "webhook" {
-
-
Webhook Configuration
-
-
- Events:
-
- if len(service.EventTriggers) == 0 {
- None
- } else {
- for i, trigger := range service.EventTriggers {
- if i > 0 {
- ,
- }
- { trigger }
- }
- }
-
-
-
- Retry:
-
- if service.RetryPolicy == "" {
- Default
- } else {
- { service.RetryPolicy }
- }
-
-
-
- Secret Key:
-
- if service.SecretKey == "" {
- None
- } else {
-
-
-
- Custom Payload:
-
- if service.PayloadTemplate == "" {
- Default
- } else {
-
-
-
-
- }
-
+ }
+
}
+
+ // if len(data.NotificationServices) == 0 {
+ //
+ //
+ //
+ //
No notification services configured
+ //
Add your first notification service to start receiving alerts about jobs and system events.
+ //
+ //
+ //
+ // } else {
+ //
+ // for _, service := range data.NotificationServices {
+ //
+ //
+ //
+ // if service.Type == "email" {
+ //
+ //
+ // } else if service.Type == "webhook" {
+ //
+ //
+ // } else {
+ //
+ //
+ // }
+ //
{ service.Name }
+ //
+ //
+ //
+ //
+ //
+ //
+ //
+ //
+ //
+ //
+ //
+ //
+ // if service.IsEnabled {
+ // Active
+ // } else {
+ // Disabled
+ // }
+ //
+ //
+ // { service.Type }
+ //
+ // if len(service.EventTriggers) > 0 && service.Type == "webhook" {
+ //
+ // { fmt.Sprintf("%d triggers", len(service.EventTriggers)) }
+ //
+ // }
+ // if service.SuccessCount > 0 || service.FailureCount > 0 {
+ //
+ // { fmt.Sprintf("%d/%d", service.SuccessCount, service.SuccessCount + service.FailureCount) }
+ //
+ // }
+ //
+ //
{ service.Description }
+ // if service.Type == "webhook" {
+ //
+ //
Webhook Configuration
+ //
+ //
+ // Events:
+ //
+ // if len(service.EventTriggers) == 0 {
+ // None
+ // } else {
+ // for i, trigger := range service.EventTriggers {
+ // if i > 0 {
+ // ,
+ // }
+ // { trigger }
+ // }
+ // }
+ //
+ //
+ //
+ // Retry:
+ //
+ // if service.RetryPolicy == "" {
+ // Default
+ // } else {
+ // { service.RetryPolicy }
+ // }
+ //
+ //
+ //
+ // Secret Key:
+ //
+ // if service.SecretKey == "" {
+ // None
+ // } else {
+ //
+ //
+ //
+ // Custom Payload:
+ //
+ // if service.PayloadTemplate == "" {
+ // Default
+ // } else {
+ //
+ //
+ //
+ //
+ // }
+ //
+ // }
+ //
+ // }
@@ -308,13 +449,12 @@ templ Settings(ctx context.Context, data SettingsData) {
-
-
Webhook URL
@@ -380,8 +509,31 @@ templ Settings(ctx context.Context, data SettingsData) {
Payload Template (JSON)
-
-
Use {`job.field`} as placeholders for job data
+
+
Use {`variable`} placeholders for dynamic values. Available variables: job.*, instance.*, timestamp, notification.*
Event Triggers
@@ -413,6 +565,27 @@ templ Settings(ctx context.Context, data SettingsData) {
Exponential backoff
+
+
+
+
+
Test Configuration
+
+
+
+
+ Send a test notification to verify your configuration works correctly before saving.
+
+
+
+
+
@@ -433,6 +606,7 @@ templ Settings(ctx context.Context, data SettingsData) {
}
+ @toggleNotificationFields()
}
script toggleNotificationFields() {
@@ -454,4 +628,4 @@ script toggleNotificationFields() {
}
});
});
-}
\ No newline at end of file
+}
\ No newline at end of file
diff --git a/internal/db/migrations/001_initial_schema.go b/internal/db/migrations/001_initial_schema.go
index 30d6a51..4030459 100644
--- a/internal/db/migrations/001_initial_schema.go
+++ b/internal/db/migrations/001_initial_schema.go
@@ -2,6 +2,9 @@ package migrations
import (
"fmt"
+ "os"
+ "path/filepath"
+ "time"
"github.com/go-gormigrate/gormigrate/v2"
"gorm.io/gorm"
@@ -11,6 +14,60 @@ func InitialSchema() *gormigrate.Migration {
return &gormigrate.Migration{
ID: "001_initial_schema",
Migrate: func(tx *gorm.DB) error {
+ // Check if any tables exist (indicating an existing database)
+ var count int64
+ if err := tx.Raw("SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").Scan(&count).Error; err != nil {
+ return fmt.Errorf("failed to check for existing tables: %v", err)
+ }
+
+ // If tables exist, create a backup
+ if count > 0 {
+ // Get the database path
+ sqlDB, err := tx.DB()
+ if err != nil {
+ return fmt.Errorf("failed to get underlying database: %v", err)
+ }
+
+ var seq int
+ var name, dbPath string
+ if err := sqlDB.QueryRow("PRAGMA database_list").Scan(&seq, &name, &dbPath); err != nil {
+ return fmt.Errorf("failed to get database path: %v", err)
+ }
+
+ // Get backup directory from environment variable or use default
+ backupDir := os.Getenv("BACKUP_DIR")
+ if backupDir == "" {
+ backupDir = "/app/backups" // Default Docker path
+ // Check if we're not in Docker
+ if _, err := os.Stat(backupDir); os.IsNotExist(err) {
+ backupDir = "backups" // Fallback to local directory
+ }
+ }
+
+ // Create backup directory if it doesn't exist
+ if err := os.MkdirAll(backupDir, 0755); err != nil {
+ return fmt.Errorf("failed to create backup directory: %v", err)
+ }
+
+ // Create backup file with timestamp in the backup directory
+ dbFileName := filepath.Base(dbPath)
+ backupFileName := fmt.Sprintf("%s.backup.%s", dbFileName, time.Now().Format("20060102_150405"))
+ backupFile := filepath.Join(backupDir, backupFileName)
+
+ // Read original database
+ data, err := os.ReadFile(dbPath)
+ if err != nil {
+ return fmt.Errorf("failed to read database for backup: %v", err)
+ }
+
+ // Write backup
+ if err := os.WriteFile(backupFile, data, 0600); err != nil {
+ return fmt.Errorf("failed to create database backup: %v", err)
+ }
+
+ fmt.Printf("Created database backup at: %s\n", backupFile)
+ }
+
// Disable foreign key constraints while creating tables
if err := tx.Exec("PRAGMA foreign_keys = OFF").Error; err != nil {
return fmt.Errorf("failed to disable foreign key constraints: %v", err)
diff --git a/internal/db/migrations/006_add_timestamps_to_job_histories.go b/internal/db/migrations/006_add_timestamps_to_job_histories.go
index e5bf802..5f27dc9 100644
--- a/internal/db/migrations/006_add_timestamps_to_job_histories.go
+++ b/internal/db/migrations/006_add_timestamps_to_job_histories.go
@@ -3,6 +3,8 @@ package migrations
import (
"encoding/json"
"fmt"
+ "os"
+ "path/filepath"
"time"
"github.com/go-gormigrate/gormigrate/v2"
@@ -14,6 +16,60 @@ func AddTimestampsToJobHistories() *gormigrate.Migration {
return &gormigrate.Migration{
ID: "006_add_timestamps_to_job_histories",
Migrate: func(tx *gorm.DB) error {
+ // Check if any tables exist (indicating an existing database)
+ var count int64
+ if err := tx.Raw("SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").Scan(&count).Error; err != nil {
+ return fmt.Errorf("failed to check for existing tables: %v", err)
+ }
+
+ // If tables exist, create a backup
+ if count > 0 {
+ // Get the database path
+ sqlDB, err := tx.DB()
+ if err != nil {
+ return fmt.Errorf("failed to get underlying database: %v", err)
+ }
+
+ var seq int
+ var name, dbPath string
+ if err := sqlDB.QueryRow("PRAGMA database_list").Scan(&seq, &name, &dbPath); err != nil {
+ return fmt.Errorf("failed to get database path: %v", err)
+ }
+
+ // Get backup directory from environment variable or use default
+ backupDir := os.Getenv("BACKUP_DIR")
+ if backupDir == "" {
+ backupDir = "/app/backups" // Default Docker path
+ // Check if we're not in Docker
+ if _, err := os.Stat(backupDir); os.IsNotExist(err) {
+ backupDir = "backups" // Fallback to local directory
+ }
+ }
+
+ // Create backup directory if it doesn't exist
+ if err := os.MkdirAll(backupDir, 0755); err != nil {
+ return fmt.Errorf("failed to create backup directory: %v", err)
+ }
+
+ // Create backup file with timestamp in the backup directory
+ dbFileName := filepath.Base(dbPath)
+ backupFileName := fmt.Sprintf("%s.backup.%s", dbFileName, time.Now().Format("20060102_150405"))
+ backupFile := filepath.Join(backupDir, backupFileName)
+
+ // Read original database
+ data, err := os.ReadFile(dbPath)
+ if err != nil {
+ return fmt.Errorf("failed to read database for backup: %v", err)
+ }
+
+ // Write backup
+ if err := os.WriteFile(backupFile, data, 0600); err != nil {
+ return fmt.Errorf("failed to create database backup: %v", err)
+ }
+
+ fmt.Printf("Created database backup at: %s\n", backupFile)
+ }
+
// Add created_at column
if err := tx.Exec("ALTER TABLE job_histories ADD COLUMN created_at DATETIME").Error; err != nil {
return fmt.Errorf("failed to add created_at column: %v", err)
diff --git a/internal/db/migrations/007_add_notification_services.go b/internal/db/migrations/007_add_notification_services.go
index f225a09..677801a 100644
--- a/internal/db/migrations/007_add_notification_services.go
+++ b/internal/db/migrations/007_add_notification_services.go
@@ -3,6 +3,8 @@ package migrations
import (
"encoding/json"
"fmt"
+ "os"
+ "path/filepath"
"time"
"github.com/go-gormigrate/gormigrate/v2"
@@ -14,6 +16,60 @@ func AddNotificationServices() *gormigrate.Migration {
return &gormigrate.Migration{
ID: "007_add_notification_services",
Migrate: func(tx *gorm.DB) error {
+ // Check if any tables exist (indicating an existing database)
+ var count int64
+ if err := tx.Raw("SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").Scan(&count).Error; err != nil {
+ return fmt.Errorf("failed to check for existing tables: %v", err)
+ }
+
+ // If tables exist, create a backup
+ if count > 0 {
+ // Get the database path
+ sqlDB, err := tx.DB()
+ if err != nil {
+ return fmt.Errorf("failed to get underlying database: %v", err)
+ }
+
+ var seq int
+ var name, dbPath string
+ if err := sqlDB.QueryRow("PRAGMA database_list").Scan(&seq, &name, &dbPath); err != nil {
+ return fmt.Errorf("failed to get database path: %v", err)
+ }
+
+ // Get backup directory from environment variable or use default
+ backupDir := os.Getenv("BACKUP_DIR")
+ if backupDir == "" {
+ backupDir = "/app/backups" // Default Docker path
+ // Check if we're not in Docker
+ if _, err := os.Stat(backupDir); os.IsNotExist(err) {
+ backupDir = "backups" // Fallback to local directory
+ }
+ }
+
+ // Create backup directory if it doesn't exist
+ if err := os.MkdirAll(backupDir, 0755); err != nil {
+ return fmt.Errorf("failed to create backup directory: %v", err)
+ }
+
+ // Create backup file with timestamp in the backup directory
+ dbFileName := filepath.Base(dbPath)
+ backupFileName := fmt.Sprintf("%s.backup.%s", dbFileName, time.Now().Format("20060102_150405"))
+ backupFile := filepath.Join(backupDir, backupFileName)
+
+ // Read original database
+ data, err := os.ReadFile(dbPath)
+ if err != nil {
+ return fmt.Errorf("failed to read database for backup: %v", err)
+ }
+
+ // Write backup
+ if err := os.WriteFile(backupFile, data, 0600); err != nil {
+ return fmt.Errorf("failed to create database backup: %v", err)
+ }
+
+ fmt.Printf("Created database backup at: %s\n", backupFile)
+ }
+
// Create notification_services table
type NotificationService struct {
ID uint `gorm:"primaryKey"`
diff --git a/internal/db/migrations/008_add_user_notifications.go b/internal/db/migrations/008_add_user_notifications.go
new file mode 100644
index 0000000..f12282e
--- /dev/null
+++ b/internal/db/migrations/008_add_user_notifications.go
@@ -0,0 +1,122 @@
+package migrations
+
+import (
+ "fmt"
+ "os"
+ "path/filepath"
+ "time"
+
+ "github.com/go-gormigrate/gormigrate/v2"
+ "gorm.io/gorm"
+)
+
+// AddUserNotifications adds the user_notifications table
+func AddUserNotifications() *gormigrate.Migration {
+ return &gormigrate.Migration{
+ ID: "008_add_user_notifications",
+ Migrate: func(tx *gorm.DB) error {
+ // Check if any tables exist (indicating an existing database)
+ var count int64
+ if err := tx.Raw("SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").Scan(&count).Error; err != nil {
+ return fmt.Errorf("failed to check for existing tables: %v", err)
+ }
+
+ // If tables exist, create a backup
+ if count > 0 {
+ // Get the database path
+ sqlDB, err := tx.DB()
+ if err != nil {
+ return fmt.Errorf("failed to get underlying database: %v", err)
+ }
+
+ var seq int
+ var name, dbPath string
+ if err := sqlDB.QueryRow("PRAGMA database_list").Scan(&seq, &name, &dbPath); err != nil {
+ return fmt.Errorf("failed to get database path: %v", err)
+ }
+
+ // Get backup directory from environment variable or use default
+ backupDir := os.Getenv("BACKUP_DIR")
+ if backupDir == "" {
+ backupDir = "/app/backups" // Default Docker path
+ // Check if we're not in Docker
+ if _, err := os.Stat(backupDir); os.IsNotExist(err) {
+ backupDir = "backups" // Fallback to local directory
+ }
+ }
+
+ // Create backup directory if it doesn't exist
+ if err := os.MkdirAll(backupDir, 0755); err != nil {
+ return fmt.Errorf("failed to create backup directory: %v", err)
+ }
+
+ // Create backup file with timestamp in the backup directory
+ dbFileName := filepath.Base(dbPath)
+ backupFileName := fmt.Sprintf("%s.backup.%s", dbFileName, time.Now().Format("20060102_150405"))
+ backupFile := filepath.Join(backupDir, backupFileName)
+
+ // Read original database
+ data, err := os.ReadFile(dbPath)
+ if err != nil {
+ return fmt.Errorf("failed to read database for backup: %v", err)
+ }
+
+ // Write backup
+ if err := os.WriteFile(backupFile, data, 0600); err != nil {
+ return fmt.Errorf("failed to create database backup: %v", err)
+ }
+
+ fmt.Printf("Created database backup at: %s\n", backupFile)
+ }
+ // Create the user_notifications table
+ err := tx.Exec(`
+ CREATE TABLE IF NOT EXISTS user_notifications (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_id INTEGER NOT NULL,
+ type TEXT NOT NULL,
+ title TEXT NOT NULL,
+ message TEXT NOT NULL,
+ link TEXT NOT NULL,
+ job_id INTEGER,
+ job_run_id INTEGER,
+ config_id INTEGER,
+ is_read BOOLEAN NOT NULL DEFAULT 0,
+ created_at DATETIME NOT NULL,
+ FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+ )
+ `).Error
+ if err != nil {
+ return err
+ }
+
+ // Create an index on user_id for faster lookups
+ err = tx.Exec(`
+ CREATE INDEX IF NOT EXISTS idx_user_notifications_user_id ON user_notifications(user_id)
+ `).Error
+ if err != nil {
+ return err
+ }
+
+ // Create an index on created_at for faster sorting
+ err = tx.Exec(`
+ CREATE INDEX IF NOT EXISTS idx_user_notifications_created_at ON user_notifications(created_at)
+ `).Error
+ if err != nil {
+ return err
+ }
+
+ // Create an index on is_read for faster filtering of unread notifications
+ err = tx.Exec(`
+ CREATE INDEX IF NOT EXISTS idx_user_notifications_is_read ON user_notifications(is_read)
+ `).Error
+ if err != nil {
+ return err
+ }
+
+ return nil
+ },
+ Rollback: func(tx *gorm.DB) error {
+ return tx.Exec(`DROP TABLE IF EXISTS user_notifications`).Error
+ },
+ }
+}
diff --git a/internal/db/migrations/migrations.go b/internal/db/migrations/migrations.go
index a82895c..4c7157c 100644
--- a/internal/db/migrations/migrations.go
+++ b/internal/db/migrations/migrations.go
@@ -18,6 +18,7 @@ func GetMigrations(db *gorm.DB) *gormigrate.Gormigrate {
AddDefaultRoles(), // 005
AddTimestampsToJobHistories(), // 006
AddNotificationServices(), // 007
+ AddUserNotifications(), // 008
)
return gormigrate.New(db, gormigrate.DefaultOptions, migrations)
diff --git a/internal/db/notification.go b/internal/db/notification.go
new file mode 100644
index 0000000..dcc0d75
--- /dev/null
+++ b/internal/db/notification.go
@@ -0,0 +1,104 @@
+package db
+
+import (
+ "encoding/json"
+ "time"
+
+ "gorm.io/gorm"
+)
+
+// NotificationService represents a notification service configuration
+type NotificationService struct {
+ ID uint `json:"id" gorm:"primaryKey"`
+ Name string `json:"name" gorm:"not null"`
+ Type string `json:"type" gorm:"not null"` // email, webhook
+ IsEnabled bool `json:"is_enabled" gorm:"default:true"`
+ Config map[string]string `json:"config" gorm:"-"`
+ ConfigJSON string `json:"-" gorm:"column:config"`
+ Description string `json:"description"`
+ EventTriggers []string `json:"event_triggers" gorm:"-"`
+ EventTriggersJSON string `json:"-" gorm:"column:event_triggers;default:'[]'"`
+ PayloadTemplate string `json:"payload_template" gorm:"column:payload_template"`
+ SecretKey string `json:"secret_key" gorm:"column:secret_key"`
+ RetryPolicy string `json:"retry_policy" gorm:"column:retry_policy;default:'simple'"`
+ LastUsed time.Time `json:"last_used" gorm:"column:last_used"`
+ SuccessCount int `json:"success_count" gorm:"column:success_count;default:0"`
+ FailureCount int `json:"failure_count" gorm:"column:failure_count;default:0"`
+ CreatedBy uint `json:"created_by"`
+ CreatedAt time.Time `json:"created_at"`
+ UpdatedAt time.Time `json:"updated_at"`
+}
+
+// BeforeSave converts Config map and EventTriggers to JSON strings for storage
+func (n *NotificationService) BeforeSave(tx *gorm.DB) error {
+ configJSON, err := json.Marshal(n.Config)
+ if err != nil {
+ return err
+ }
+ n.ConfigJSON = string(configJSON)
+
+ eventsJSON, err := json.Marshal(n.EventTriggers)
+ if err != nil {
+ return err
+ }
+ n.EventTriggersJSON = string(eventsJSON)
+
+ return nil
+}
+
+// AfterFind converts JSON strings back to Config map and EventTriggers
+func (n *NotificationService) AfterFind(tx *gorm.DB) error {
+ if n.ConfigJSON != "" {
+ if err := json.Unmarshal([]byte(n.ConfigJSON), &n.Config); err != nil {
+ return err
+ }
+ }
+
+ if n.EventTriggersJSON != "" {
+ if err := json.Unmarshal([]byte(n.EventTriggersJSON), &n.EventTriggers); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
+// GetNotificationServices returns notification services, filtered by enabled status if specified
+func (db *DB) GetNotificationServices(onlyEnabled bool) ([]NotificationService, error) {
+ var services []NotificationService
+ query := db.DB
+
+ if onlyEnabled {
+ query = query.Where("is_enabled = ?", true)
+ }
+
+ if err := query.Find(&services).Error; err != nil {
+ return nil, err
+ }
+
+ return services, nil
+}
+
+// GetNotificationService returns a notification service by ID
+func (db *DB) GetNotificationService(id uint) (*NotificationService, error) {
+ var service NotificationService
+ if err := db.First(&service, id).Error; err != nil {
+ return nil, err
+ }
+ return &service, nil
+}
+
+// CreateNotificationService creates a new notification service
+func (db *DB) CreateNotificationService(service *NotificationService) error {
+ return db.Create(service).Error
+}
+
+// UpdateNotificationService updates an existing notification service
+func (db *DB) UpdateNotificationService(service *NotificationService) error {
+ return db.Save(service).Error
+}
+
+// DeleteNotificationService deletes a notification service by ID
+func (db *DB) DeleteNotificationService(id uint) error {
+ return db.Delete(&NotificationService{}, id).Error
+}
diff --git a/internal/db/user_notification.go b/internal/db/user_notification.go
new file mode 100644
index 0000000..0950d3d
--- /dev/null
+++ b/internal/db/user_notification.go
@@ -0,0 +1,133 @@
+package db
+
+import (
+ "fmt"
+ "time"
+)
+
+// NotificationType defines the type of notification
+type NotificationType string
+
+const (
+ NotificationJobStart NotificationType = "job_start"
+ NotificationJobComplete NotificationType = "job_complete"
+ NotificationJobFail NotificationType = "job_fail"
+ NotificationConfigUpdate NotificationType = "config_update"
+ NotificationSystemAlert NotificationType = "system_alert"
+)
+
+// UserNotification represents a notification shown to users in the UI
+type UserNotification struct {
+ ID uint `json:"id" gorm:"primaryKey"`
+ UserID uint `json:"user_id" gorm:"index"`
+ Type NotificationType `json:"type"`
+ Title string `json:"title"`
+ Message string `json:"message"`
+ Link string `json:"link"`
+ JobID uint `json:"job_id,omitempty"`
+ JobRunID uint `json:"job_run_id,omitempty"`
+ ConfigID uint `json:"config_id,omitempty"`
+ IsRead bool `json:"is_read" gorm:"default:false"`
+ CreatedAt time.Time `json:"created_at"`
+}
+
+// GetUserNotifications returns the latest notifications for a user
+func (db *DB) GetUserNotifications(userID uint, limit int) ([]UserNotification, error) {
+ var notifications []UserNotification
+ result := db.Where("user_id = ?", userID).Order("created_at DESC").Limit(limit).Find(¬ifications)
+ return notifications, result.Error
+}
+
+// GetUnreadNotificationCount returns the count of unread notifications for a user
+func (db *DB) GetUnreadNotificationCount(userID uint) (int64, error) {
+ var count int64
+ result := db.Model(&UserNotification{}).Where("user_id = ? AND is_read = ?", userID, false).Count(&count)
+ return count, result.Error
+}
+
+// MarkNotificationAsRead marks a notification as read
+func (db *DB) MarkNotificationAsRead(id uint) error {
+ return db.Model(&UserNotification{}).Where("id = ?", id).Update("is_read", true).Error
+}
+
+// MarkAllNotificationsAsRead marks all notifications for a user as read
+func (db *DB) MarkAllNotificationsAsRead(userID uint) error {
+ return db.Model(&UserNotification{}).Where("user_id = ?", userID).Update("is_read", true).Error
+}
+
+// CreateJobNotification creates a notification for a job event
+func (db *DB) CreateJobNotification(
+ userID uint,
+ jobID uint,
+ jobRunID uint,
+ notificationType NotificationType,
+ title string,
+ message string,
+) error {
+ notification := UserNotification{
+ UserID: userID,
+ Type: notificationType,
+ Title: title,
+ Message: message,
+ JobID: jobID,
+ JobRunID: jobRunID,
+ Link: generateJobRunLink(jobRunID),
+ CreatedAt: time.Now(),
+ }
+ return db.Create(¬ification).Error
+}
+
+// CreateConfigNotification creates a notification for a config update
+func (db *DB) CreateConfigNotification(
+ userID uint,
+ configID uint,
+ title string,
+ message string,
+) error {
+ notification := UserNotification{
+ UserID: userID,
+ Type: NotificationConfigUpdate,
+ Title: title,
+ Message: message,
+ ConfigID: configID,
+ Link: generateConfigLink(configID),
+ CreatedAt: time.Now(),
+ }
+ return db.Create(¬ification).Error
+}
+
+// CreateSystemNotification creates a system-wide notification
+func (db *DB) CreateSystemNotification(
+ title string,
+ message string,
+) error {
+ // Get all active users
+ var users []User
+ if err := db.Where("active = ?", true).Find(&users).Error; err != nil {
+ return err
+ }
+
+ // Create a notification for each user
+ for _, user := range users {
+ notification := UserNotification{
+ UserID: user.ID,
+ Type: NotificationSystemAlert,
+ Title: title,
+ Message: message,
+ CreatedAt: time.Now(),
+ }
+ if err := db.Create(¬ification).Error; err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// Helper functions to generate links
+func generateJobRunLink(jobRunID uint) string {
+ return "/job-runs/" + fmt.Sprintf("%d", jobRunID)
+}
+
+func generateConfigLink(configID uint) string {
+ return "/configs/" + fmt.Sprintf("%d", configID)
+}
diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 2a06228..dc4c6e8 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -447,6 +447,9 @@ func (s *Scheduler) processConfiguration(job *db.Job, config *db.TransferConfig,
s.log.LogDebug("Creating job history record: %+v", history)
+ // Send webhook notification for job start
+ s.sendWebhookNotification(job, history, config)
+
// Execute the configuration transfer
s.executeConfigTransfer(*job, *config, history)
}
@@ -484,6 +487,7 @@ func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig,
}
// Send webhook notification for failure
s.sendWebhookNotification(&job, history, &config)
+
return
}
defer os.Remove(filterFile)
@@ -974,8 +978,14 @@ func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig,
s.log.LogError("Error updating job history for job %d, config %d: %v", job.ID, config.ID, err)
}
+ // Create job notification
+ if err := s.createJobNotification(&job, history); err != nil {
+ s.log.LogError("Failed to create job notification", "jobID", job.ID, "error", err)
+ }
+
// Send webhook notification for success or with errors
s.sendWebhookNotification(&job, history, &config)
+
}
// ProcessOutputPattern processes an output pattern with variables and returns the result
@@ -1098,20 +1108,25 @@ func (s *Scheduler) checkFileProcessingHistory(jobID uint, fileName string) (*db
// sendWebhookNotification sends a notification to the configured webhook URL
func (s *Scheduler) sendWebhookNotification(job *db.Job, history *db.JobHistory, config *db.TransferConfig) {
- if !job.GetWebhookEnabled() || job.WebhookURL == "" {
- return
+ // First, handle job-specific webhook if configured
+ if job.GetWebhookEnabled() && job.WebhookURL != "" {
+ // Skip notifications based on settings
+ if history.Status == "completed" && !job.GetNotifyOnSuccess() {
+ s.log.LogDebug("Skipping success notification for job %d (notifyOnSuccess=false)", job.ID)
+ } else if history.Status == "failed" && !job.GetNotifyOnFailure() {
+ s.log.LogDebug("Skipping failure notification for job %d (notifyOnFailure=false)", job.ID)
+ } else {
+ s.log.LogInfo("Sending job-specific webhook notification for job %d", job.ID)
+ s.sendJobWebhookNotification(job, history, config)
+ }
}
- // Skip notifications based on settings
- if history.Status == "completed" && !job.GetNotifyOnSuccess() {
- return
- }
- if history.Status == "failed" && !job.GetNotifyOnFailure() {
- return
- }
-
- s.log.LogInfo("Sending webhook notification for job %d", job.ID)
+ // Next, process global notification services
+ s.sendGlobalNotifications(job, history, config)
+}
+// sendJobWebhookNotification sends a notification to the job's configured webhook URL
+func (s *Scheduler) sendJobWebhookNotification(job *db.Job, history *db.JobHistory, config *db.TransferConfig) {
// Create the payload with useful information
payload := map[string]interface{}{
"event_type": "job_execution",
@@ -1208,3 +1223,568 @@ func (s *Scheduler) sendWebhookNotification(job *db.Job, history *db.JobHistory,
}
}
}
+
+// sendGlobalNotifications sends notifications through all configured notification services
+func (s *Scheduler) sendGlobalNotifications(job *db.Job, history *db.JobHistory, config *db.TransferConfig) {
+ // Fetch all enabled notification services
+ services, err := s.db.GetNotificationServices(true)
+ if err != nil {
+ s.log.LogError("Error fetching notification services: %v", err)
+ return
+ }
+
+ if len(services) == 0 {
+ s.log.LogDebug("No enabled notification services found")
+ return
+ }
+
+ s.log.LogInfo("Found %d enabled notification services", len(services))
+
+ // Determine event type based on job status
+ var eventType string
+ switch history.Status {
+ case "running":
+ eventType = "job_start"
+ case "completed", "completed_with_errors":
+ eventType = "job_complete"
+ case "failed":
+ eventType = "job_error"
+ default:
+ eventType = "job_status"
+ }
+
+ // Process each notification service
+ for i := range services {
+ service := &services[i] // Use pointer to update stats
+ s.log.LogInfo("Processing notification service %s (%s)", service.Name, service.Type)
+ // Check if this service should handle this event type
+ shouldSend := false
+ for _, trigger := range service.EventTriggers {
+ if trigger == eventType {
+ shouldSend = true
+ break
+ }
+ }
+
+ // Skip if this service doesn't handle this event type
+ if !shouldSend {
+ s.log.LogDebug("Skipping notification service %s (%s) for event %s (not in triggers)",
+ service.Name, service.Type, eventType)
+ continue
+ }
+
+ s.log.LogInfo("Sending notification via service %s (%s) for job %d",
+ service.Name, service.Type, job.ID)
+
+ // Send notification based on service type
+ var notifyErr error
+ switch service.Type {
+ case "email":
+ notifyErr = s.sendEmailNotification(service, job, history, config, eventType)
+ case "webhook":
+ notifyErr = s.sendServiceWebhookNotification(service, job, history, config, eventType)
+ default:
+ s.log.LogError("Unsupported notification service type: %s", service.Type)
+ continue
+ }
+
+ // Update service success/failure count
+ if notifyErr != nil {
+ service.FailureCount++
+ s.log.LogError("Notification service %s failed: %v", service.Name, notifyErr)
+ } else {
+ service.SuccessCount++
+ service.LastUsed = time.Now()
+ s.log.LogInfo("Notification service %s sent successfully", service.Name)
+ }
+
+ // Update notification service stats in the database
+ if err := s.db.UpdateNotificationService(service); err != nil {
+ s.log.LogError("Error updating notification service stats: %v", err)
+ }
+ }
+}
+
+// sendEmailNotification sends an email notification using the configured email service
+func (s *Scheduler) sendEmailNotification(service *db.NotificationService, job *db.Job, history *db.JobHistory, config *db.TransferConfig, eventType string) error {
+ s.log.LogDebug("Preparing email notification via service %s for job %d", service.Name, job.ID)
+
+ // Extract SMTP settings from service config
+ smtpHost := service.Config["smtp_host"]
+ smtpPortStr := service.Config["smtp_port"]
+ fromEmail := service.Config["from_email"]
+ toEmail := service.Config["to_email"]
+
+ // Validate required settings
+ if smtpHost == "" || smtpPortStr == "" || fromEmail == "" || toEmail == "" {
+ return fmt.Errorf("missing required SMTP settings")
+ }
+
+ // Parse SMTP port
+ smtpPort, err := strconv.Atoi(smtpPortStr)
+ if err != nil {
+ return fmt.Errorf("invalid SMTP port: %v", err)
+ }
+
+ // Prepare email content
+ subject := fmt.Sprintf("[GoMFT] Job %s: %s", job.Name, history.Status)
+ body := generateEmailBody(job, history, config, eventType)
+
+ // TODO: Implement actual email sending logic
+ // This would typically involve using a package like "net/smtp" or a third-party
+ // email library to send the actual email.
+ // For actual implementation, you would use:
+ // - smtpUsername := service.Config["smtp_username"]
+ // - smtpPassword := service.Config["smtp_password"]
+
+ s.log.LogInfo("Email would be sent to %s with subject: %s", toEmail, subject)
+ s.log.LogDebug("Email body: %s", body)
+
+ // Placeholder for actual email sending
+ // For now, we'll just log that the email would be sent
+ s.log.LogInfo("Email notification prepared (SMTP: %s:%d, From: %s, To: %s)",
+ smtpHost, smtpPort, fromEmail, toEmail)
+
+ return nil
+}
+
+// generateEmailBody creates the email body for job notifications
+func generateEmailBody(job *db.Job, history *db.JobHistory, config *db.TransferConfig, eventType string) string {
+ var b strings.Builder
+
+ b.WriteString(fmt.Sprintf("Job: %s (ID: %d)\n", job.Name, job.ID))
+ b.WriteString(fmt.Sprintf("Status: %s\n", history.Status))
+ b.WriteString(fmt.Sprintf("Start Time: %s\n", history.StartTime.Format(time.RFC3339)))
+
+ if history.EndTime != nil {
+ b.WriteString(fmt.Sprintf("End Time: %s\n", history.EndTime.Format(time.RFC3339)))
+ duration := history.EndTime.Sub(history.StartTime)
+ b.WriteString(fmt.Sprintf("Duration: %.2f seconds\n", duration.Seconds()))
+ }
+
+ b.WriteString(fmt.Sprintf("Files Transferred: %d\n", history.FilesTransferred))
+ b.WriteString(fmt.Sprintf("Bytes Transferred: %d\n", history.BytesTransferred))
+
+ b.WriteString("\nTransfer Configuration:\n")
+ b.WriteString(fmt.Sprintf("Name: %s (ID: %d)\n", config.Name, config.ID))
+ b.WriteString(fmt.Sprintf("Source: %s:%s\n", config.SourceType, config.SourcePath))
+ b.WriteString(fmt.Sprintf("Destination: %s:%s\n", config.DestinationType, config.DestinationPath))
+
+ if history.ErrorMessage != "" {
+ b.WriteString("\nError Details:\n")
+ b.WriteString(history.ErrorMessage)
+ }
+
+ return b.String()
+}
+
+// sendServiceWebhookNotification sends a webhook notification using a configured notification service
+func (s *Scheduler) sendServiceWebhookNotification(service *db.NotificationService, job *db.Job, history *db.JobHistory, config *db.TransferConfig, eventType string) error {
+ s.log.LogDebug("Preparing webhook notification via service %s for job %d", service.Name, job.ID)
+
+ // Extract webhook settings
+ webhookURL := service.Config["webhook_url"]
+ method := service.Config["method"]
+ if method == "" {
+ method = "POST" // Default to POST if not specified
+ }
+
+ // Validate required settings
+ if webhookURL == "" {
+ return fmt.Errorf("missing webhook URL")
+ }
+
+ // Prepare payload
+ var payload map[string]interface{}
+
+ // Use custom payload template if provided
+ if service.PayloadTemplate != "" {
+ // Parse the template and fill in variables
+ payload = generateCustomPayload(service.PayloadTemplate, job, history, config, eventType)
+ } else {
+ // Use default payload format
+ payload = generateDefaultPayload(job, history, config, eventType)
+ }
+
+ // Convert payload to JSON
+ jsonPayload, err := json.Marshal(payload)
+ if err != nil {
+ return fmt.Errorf("error marshaling webhook payload: %v", err)
+ }
+
+ s.log.LogDebug("Webhook payload: %s", string(jsonPayload))
+
+ // Create HTTP request
+ req, err := http.NewRequest(method, webhookURL, bytes.NewBuffer(jsonPayload))
+ if err != nil {
+ return fmt.Errorf("error creating webhook request: %v", err)
+ }
+
+ // Set default headers
+ req.Header.Set("Content-Type", "application/json")
+ req.Header.Set("User-Agent", "GoMFT-Notification/1.0")
+
+ // Add signature if secret key is provided
+ if service.SecretKey != "" {
+ h := hmac.New(sha256.New, []byte(service.SecretKey))
+ h.Write(jsonPayload)
+ signature := hex.EncodeToString(h.Sum(nil))
+ req.Header.Set("X-GoMFT-Signature", signature)
+ }
+
+ // Add custom headers if specified
+ if headersStr := service.Config["headers"]; headersStr != "" {
+ var headers map[string]string
+ if err := json.Unmarshal([]byte(headersStr), &headers); err == nil {
+ for key, value := range headers {
+ req.Header.Set(key, value)
+ }
+ }
+ }
+
+ s.log.LogDebug("Webhook headers: %+v", req.Header)
+
+ // Determine timeout based on retry policy
+ timeout := 10 * time.Second
+ maxRetries := 0
+
+ switch service.RetryPolicy {
+ case "none":
+ maxRetries = 0
+ case "simple":
+ maxRetries = 3
+ timeout = 15 * time.Second
+ case "exponential":
+ maxRetries = 5
+ timeout = 30 * time.Second
+ default:
+ // Default to simple
+ maxRetries = 3
+ timeout = 15 * time.Second
+ }
+
+ // Prepare client with timeout
+ client := &http.Client{
+ Timeout: timeout,
+ }
+
+ // Attempt to send with retries
+ var resp *http.Response
+ var lastErr error
+
+ for attempt := 0; attempt <= maxRetries; attempt++ {
+ if attempt > 0 {
+ // Wait before retry with increasing backoff
+ backoffDuration := time.Duration(1<
= 200 && resp.StatusCode < 300 {
+ defer resp.Body.Close()
+ s.log.LogInfo("Webhook notification sent successfully (status: %d)", resp.StatusCode)
+ return nil
+ }
+
+ // Error status code
+ respBody, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ lastErr = fmt.Errorf("webhook returned status %d: %s", resp.StatusCode, respBody)
+ s.log.LogError("Webhook error (attempt %d/%d): %v", attempt+1, maxRetries+1, lastErr)
+ } else {
+ // Network or request error
+ lastErr = fmt.Errorf("webhook request failed: %v", err)
+ s.log.LogError("Webhook request error (attempt %d/%d): %v", attempt+1, maxRetries+1, lastErr)
+ }
+ }
+
+ return lastErr
+}
+
+// generateDefaultPayload creates a standard webhook payload
+func generateDefaultPayload(job *db.Job, history *db.JobHistory, config *db.TransferConfig, eventType string) map[string]interface{} {
+ payload := map[string]interface{}{
+ "event": eventType,
+ "job": map[string]interface{}{
+ "id": job.ID,
+ "name": job.Name,
+ "status": history.Status,
+ "message": history.ErrorMessage,
+ "started_at": history.StartTime.Format(time.RFC3339),
+ "config_id": config.ID,
+ "config_name": config.Name,
+ "transfer_bytes": history.BytesTransferred,
+ "file_count": history.FilesTransferred,
+ },
+ "instance": map[string]interface{}{
+ "id": "gomft",
+ "name": "GoMFT",
+ "version": "1.0", // TODO: Get actual version
+ "environment": "production", // TODO: Get from env
+ },
+ "timestamp": time.Now().Format(time.RFC3339),
+ }
+
+ if history.EndTime != nil {
+ payload["job"].(map[string]interface{})["completed_at"] = history.EndTime.Format(time.RFC3339)
+ duration := history.EndTime.Sub(history.StartTime)
+ payload["job"].(map[string]interface{})["duration_seconds"] = duration.Seconds()
+ }
+
+ return payload
+}
+
+// generateCustomPayload creates a webhook payload from a template
+func generateCustomPayload(template string, job *db.Job, history *db.JobHistory, config *db.TransferConfig, eventType string) map[string]interface{} {
+ // Start with the default payload as a base
+ defaultPayload := generateDefaultPayload(job, history, config, eventType)
+
+ // Parse the template string to JSON
+ var customPayload map[string]interface{}
+ if err := json.Unmarshal([]byte(template), &customPayload); err != nil {
+ // If template can't be parsed, fall back to default payload
+ return defaultPayload
+ }
+
+ // Replace variables in the template
+ // This is a simplified version - a real implementation would do deep traversal
+ // and replace all variables in the structure
+ processedPayload := processPayloadVariables(customPayload, defaultPayload)
+
+ return processedPayload
+}
+
+// processPayloadVariables recursively processes a payload structure and replaces variables
+func processPayloadVariables(customPayload map[string]interface{}, variables map[string]interface{}) map[string]interface{} {
+ result := make(map[string]interface{})
+
+ // Process each key-value pair in the custom payload
+ for key, value := range customPayload {
+ switch v := value.(type) {
+ case string:
+ // Replace string variables
+ result[key] = replaceVariables(v, variables)
+ case map[string]interface{}:
+ // Recursively process nested maps
+ result[key] = processPayloadVariables(v, variables)
+ case []interface{}:
+ // Process arrays
+ result[key] = processArrayVariables(v, variables)
+ default:
+ // Keep other types as is
+ result[key] = value
+ }
+ }
+
+ return result
+}
+
+// processArrayVariables processes array elements for variable replacement
+func processArrayVariables(array []interface{}, variables map[string]interface{}) []interface{} {
+ result := make([]interface{}, len(array))
+
+ for i, value := range array {
+ switch v := value.(type) {
+ case string:
+ result[i] = replaceVariables(v, variables)
+ case map[string]interface{}:
+ result[i] = processPayloadVariables(v, variables)
+ case []interface{}:
+ result[i] = processArrayVariables(v, variables)
+ default:
+ result[i] = value
+ }
+ }
+
+ return result
+}
+
+// replaceVariables replaces variable placeholders in a string with their values
+func replaceVariables(template string, variables map[string]interface{}) string {
+ // Check for variable pattern like {{job.name}}
+ re := regexp.MustCompile(`{{([^{}]+)}}`)
+ result := re.ReplaceAllStringFunc(template, func(match string) string {
+ // Extract variable path (e.g., "job.name")
+ varPath := re.FindStringSubmatch(match)[1]
+ parts := strings.Split(varPath, ".")
+
+ // Navigate the variables structure to find the value
+ var current interface{} = variables
+ for _, part := range parts {
+ if m, ok := current.(map[string]interface{}); ok {
+ if val, exists := m[part]; exists {
+ current = val
+ } else {
+ return match // Keep original if not found
+ }
+ } else {
+ return match // Keep original if structure doesn't match
+ }
+ }
+
+ // Convert the found value to string
+ switch v := current.(type) {
+ case string:
+ return v
+ case int, int64, uint, uint64, float32, float64:
+ return fmt.Sprintf("%v", v)
+ case bool:
+ return fmt.Sprintf("%v", v)
+ case time.Time:
+ return v.Format(time.RFC3339)
+ default:
+ // For complex types, convert to JSON
+ if bytes, err := json.Marshal(v); err == nil {
+ return string(bytes)
+ }
+ return match
+ }
+ })
+
+ return result
+}
+
+func (s *Scheduler) updateJobStatus(jobID uint, status string, startTime, endTime time.Time, message string) (*db.JobHistory, error) {
+ // Create the history record
+ history := &db.JobHistory{
+ JobID: jobID,
+ Status: status,
+ StartTime: startTime,
+ EndTime: &endTime,
+ ErrorMessage: message,
+ }
+
+ // Add code to create notifications for job events
+ if job, err := s.db.GetJob(jobID); err == nil {
+ // Get the user who created the job
+ userID := job.CreatedBy
+
+ // Create job title from job name or ID
+ jobTitle := job.Name
+ if jobTitle == "" {
+ jobTitle = fmt.Sprintf("Job #%d", job.ID)
+ }
+
+ // Check which notification to send based on status
+ var notificationType db.NotificationType
+ var title string
+ var message string
+
+ switch status {
+ case "running":
+ notificationType = db.NotificationJobStart
+ title = "Job Started"
+ message = jobTitle
+ case "completed":
+ notificationType = db.NotificationJobComplete
+ title = "Job Complete"
+ message = jobTitle
+ case "failed":
+ notificationType = db.NotificationJobFail
+ title = "Job Failed"
+ message = jobTitle
+ if history.ErrorMessage != "" {
+ message = jobTitle + ": " + history.ErrorMessage
+ }
+ default:
+ // Don't create notifications for other statuses
+ return history, nil
+ }
+
+ // Create the notification
+ err = s.db.CreateJobNotification(
+ userID,
+ jobID,
+ history.ID, // Use the job history ID
+ notificationType,
+ title,
+ message,
+ )
+
+ if err != nil {
+ s.log.LogError("Failed to create job notification", "jobID", job.ID, "error", err)
+ // Continue anyway, not critical
+ }
+ }
+
+ return history, nil
+}
+
+// Create a job history record and send notification
+func (s *Scheduler) createJobHistoryAndNotify(job *db.Job, status string, startTime time.Time, endTime time.Time, message string) error {
+ // Create the job history entry
+ history := db.JobHistory{
+ JobID: job.ID,
+ Status: status,
+ StartTime: startTime,
+ EndTime: &endTime,
+ ErrorMessage: message,
+ }
+
+ // Save to database
+ if err := s.db.Create(&history).Error; err != nil {
+ s.log.LogError("Failed to create job history", "jobID", job.ID, "error", err)
+ return err
+ }
+
+ // Create notification
+ err := s.createJobNotification(job, &history)
+ if err != nil {
+ s.log.LogError("Failed to create job notification", "jobID", job.ID, "error", err)
+ // Continue anyway - notification is not critical
+ }
+
+ return nil
+}
+
+// Create a notification for a job event
+func (s *Scheduler) createJobNotification(job *db.Job, history *db.JobHistory) error {
+ // Get the user who created the job
+ userID := job.CreatedBy
+
+ // Create job title from job name or ID
+ jobTitle := job.Name
+ if jobTitle == "" {
+ jobTitle = fmt.Sprintf("Job #%d", job.ID)
+ }
+
+ // Determine notification type and content
+ var notificationType db.NotificationType
+ var title string
+ var message string
+
+ switch history.Status {
+ case "running":
+ notificationType = db.NotificationJobStart
+ title = "Job Started"
+ message = jobTitle
+ case "completed":
+ notificationType = db.NotificationJobComplete
+ title = "Job Complete"
+ message = jobTitle
+ case "failed":
+ notificationType = db.NotificationJobFail
+ title = "Job Failed"
+ message = jobTitle
+ if history.ErrorMessage != "" {
+ message = jobTitle + ": " + history.ErrorMessage
+ }
+ default:
+ // Don't create notifications for other statuses
+ return nil
+ }
+
+ // Create the notification
+ return s.db.CreateJobNotification(
+ userID,
+ job.ID,
+ history.ID,
+ notificationType,
+ title,
+ message,
+ )
+}
diff --git a/internal/web/handlers/config_handlers.go b/internal/web/handlers/config_handlers.go
index 2370ea7..2c433b2 100644
--- a/internal/web/handlers/config_handlers.go
+++ b/internal/web/handlers/config_handlers.go
@@ -427,3 +427,145 @@ func (h *Handlers) HandleDeleteConfig(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"message": "Config deleted successfully"})
}
+
+// HandleDuplicateConfig handles the POST /configs/:id/duplicate route
+func (h *Handlers) HandleDuplicateConfig(c *gin.Context) {
+ id := c.Param("id")
+ userID := c.GetUint("userID")
+
+ var originalConfig db.TransferConfig
+ if err := h.DB.First(&originalConfig, id).Error; err != nil {
+ c.JSON(http.StatusNotFound, gin.H{"error": "Config not found"})
+ return
+ }
+
+ // Check if user owns this config
+ if originalConfig.CreatedBy != userID {
+ // Check if user is admin
+ isAdmin, exists := c.Get("isAdmin")
+ if !exists || isAdmin != true {
+ c.JSON(http.StatusForbidden, gin.H{"error": "You do not have permission to duplicate this config"})
+ return
+ }
+ }
+
+ // Create a duplicate config
+ duplicateConfig := originalConfig
+ duplicateConfig.ID = 0 // Set ID to 0 to create a new record
+ duplicateConfig.Name = originalConfig.Name + " - Copy"
+ duplicateConfig.CreatedAt = time.Now()
+ duplicateConfig.UpdatedAt = time.Now()
+ duplicateConfig.CreatedBy = userID
+
+ // Deep copy all boolean pointers
+ skipProcessedVal := *originalConfig.SkipProcessedFiles
+ duplicateConfig.SkipProcessedFiles = &skipProcessedVal
+
+ archiveEnabledVal := *originalConfig.ArchiveEnabled
+ duplicateConfig.ArchiveEnabled = &archiveEnabledVal
+
+ deleteAfterTransferVal := *originalConfig.DeleteAfterTransfer
+ duplicateConfig.DeleteAfterTransfer = &deleteAfterTransferVal
+
+ sourcePassiveModeVal := *originalConfig.SourcePassiveMode
+ duplicateConfig.SourcePassiveMode = &sourcePassiveModeVal
+
+ destPassiveModeVal := *originalConfig.DestPassiveMode
+ duplicateConfig.DestPassiveMode = &destPassiveModeVal
+
+ // Google Photos specific fields
+ if originalConfig.DestReadOnly != nil {
+ destReadOnlyVal := *originalConfig.DestReadOnly
+ duplicateConfig.DestReadOnly = &destReadOnlyVal
+ }
+
+ if originalConfig.SourceReadOnly != nil {
+ sourceReadOnlyVal := *originalConfig.SourceReadOnly
+ duplicateConfig.SourceReadOnly = &sourceReadOnlyVal
+ }
+
+ if originalConfig.DestIncludeArchived != nil {
+ destIncludeArchivedVal := *originalConfig.DestIncludeArchived
+ duplicateConfig.DestIncludeArchived = &destIncludeArchivedVal
+ }
+
+ if originalConfig.SourceIncludeArchived != nil {
+ sourceIncludeArchivedVal := *originalConfig.SourceIncludeArchived
+ duplicateConfig.SourceIncludeArchived = &sourceIncludeArchivedVal
+ }
+
+ if originalConfig.UseBuiltinAuthSource != nil {
+ useBuiltinAuthSourceVal := *originalConfig.UseBuiltinAuthSource
+ duplicateConfig.UseBuiltinAuthSource = &useBuiltinAuthSourceVal
+ }
+
+ if originalConfig.UseBuiltinAuthDest != nil {
+ useBuiltinAuthDestVal := *originalConfig.UseBuiltinAuthDest
+ duplicateConfig.UseBuiltinAuthDest = &useBuiltinAuthDestVal
+ }
+
+ // Start a transaction
+ tx := h.DB.Begin()
+ if tx.Error != nil {
+ log.Printf("Error beginning transaction: %v", tx.Error)
+ c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to begin transaction"})
+ return
+ }
+
+ if err := tx.Create(&duplicateConfig).Error; err != nil {
+ tx.Rollback()
+ log.Printf("Error creating duplicate config: %v", err)
+ c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("Failed to create duplicate config: %v", err)})
+ return
+ }
+
+ // Create audit log entry
+ auditDetails := map[string]interface{}{
+ "name": duplicateConfig.Name,
+ "source_type": duplicateConfig.SourceType,
+ "dest_type": duplicateConfig.DestinationType,
+ "source_path": duplicateConfig.SourcePath,
+ "dest_path": duplicateConfig.DestinationPath,
+ "skip_processed_files": *duplicateConfig.SkipProcessedFiles,
+ "archive_enabled": *duplicateConfig.ArchiveEnabled,
+ "delete_after_transfer": *duplicateConfig.DeleteAfterTransfer,
+ "source_passive_mode": *duplicateConfig.SourcePassiveMode,
+ "dest_passive_mode": *duplicateConfig.DestPassiveMode,
+ "duplicated_from": originalConfig.ID,
+ }
+
+ auditLog := db.AuditLog{
+ Action: "duplicate",
+ EntityType: "config",
+ EntityID: duplicateConfig.ID,
+ UserID: userID,
+ Details: auditDetails,
+ Timestamp: time.Now(),
+ }
+
+ if err := tx.Create(&auditLog).Error; err != nil {
+ tx.Rollback()
+ log.Printf("Error creating audit log: %v", err)
+ c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create audit log"})
+ return
+ }
+
+ // Commit the transaction
+ if err := tx.Commit().Error; err != nil {
+ log.Printf("Error committing transaction: %v", err)
+ c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to commit transaction"})
+ return
+ }
+
+ // Generate rclone config file for the duplicate
+ if err := h.DB.GenerateRcloneConfig(&duplicateConfig); err != nil {
+ log.Printf("Warning: Failed to generate rclone config for duplicate: %v", err)
+ // Continue anyway, as the config was created in the database
+ } else {
+ log.Printf("Generated rclone config for duplicate config ID %d", duplicateConfig.ID)
+ }
+
+ // Return with full page reload to show the new config
+ c.Header("HX-Refresh", "true")
+ c.JSON(http.StatusOK, gin.H{"message": "Config duplicated successfully"})
+}
diff --git a/internal/web/handlers/job_handlers.go b/internal/web/handlers/job_handlers.go
index 2d9ab8b..6ac8cfd 100644
--- a/internal/web/handlers/job_handlers.go
+++ b/internal/web/handlers/job_handlers.go
@@ -733,3 +733,87 @@ func (h *Handlers) HandleRunJob(c *gin.Context) {
successScript := fmt.Sprintf("", jobName)
c.String(http.StatusOK, successScript)
}
+
+// HandleDuplicateJob handles duplication of a job
+func (h *Handlers) HandleDuplicateJob(c *gin.Context) {
+ // Get the job ID from the URL
+ idParam := c.Param("id")
+ id, err := strconv.ParseUint(idParam, 10, 64)
+ if err != nil {
+ c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid job ID"})
+ return
+ }
+
+ userID := c.GetUint("userID")
+
+ // Get the original job
+ var originalJob db.Job
+ if err := h.DB.First(&originalJob, id).Error; err != nil {
+ c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"})
+ return
+ }
+
+ // Create a new job as a copy of the original
+ newJob := originalJob
+ newJob.ID = 0 // Reset ID to create a new record
+ newJob.Name = originalJob.Name + " - Copy"
+ newJob.CreatedAt = time.Now()
+ newJob.UpdatedAt = time.Now()
+
+ // Reset execution specific fields
+ newJob.LastRun = nil
+ newJob.NextRun = nil
+
+ // Save the new job
+ if err := h.DB.Create(&newJob).Error; err != nil {
+ c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create duplicate job: " + err.Error()})
+ return
+ }
+
+ // If the job has associated configs, duplicate those associations
+ configIDs := originalJob.GetConfigIDsList()
+ if len(configIDs) > 0 {
+ // Set the new job's config IDs
+ newJob.SetConfigIDsList(configIDs)
+ if err := h.DB.Save(&newJob).Error; err != nil {
+ c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update config associations: " + err.Error()})
+ return
+ }
+ }
+
+ // Create audit log entry
+ auditDetails := map[string]interface{}{
+ "name": newJob.Name,
+ "original_job_id": originalJob.ID,
+ "new_job_id": newJob.ID,
+ "schedule": newJob.Schedule,
+ "enabled": newJob.GetEnabled(),
+ "config_ids": configIDs,
+ "webhook_enabled": newJob.GetWebhookEnabled(),
+ "notify_on_success": newJob.GetNotifyOnSuccess(),
+ "notify_on_failure": newJob.GetNotifyOnFailure(),
+ }
+
+ auditLog := db.AuditLog{
+ Action: "duplicate",
+ EntityType: "job",
+ EntityID: newJob.ID,
+ UserID: userID,
+ Details: auditDetails,
+ Timestamp: time.Now(),
+ }
+
+ if err := h.DB.Create(&auditLog).Error; err != nil {
+ log.Printf("Warning: Failed to create audit log for job duplication: %v", err)
+ // Continue anyway, as this is not critical
+ }
+
+ // Schedule the job with the scheduler
+ if err := h.Scheduler.ScheduleJob(&newJob); err != nil {
+ log.Printf("Warning: Failed to schedule duplicated job: %v", err)
+ // Continue anyway, as user can manually schedule later
+ }
+
+ // Redirect to jobs page
+ c.Redirect(http.StatusFound, "/jobs")
+}
diff --git a/internal/web/handlers/notifications_handlers.go b/internal/web/handlers/notifications_handlers.go
new file mode 100644
index 0000000..57638b5
--- /dev/null
+++ b/internal/web/handlers/notifications_handlers.go
@@ -0,0 +1,122 @@
+package handlers
+
+import (
+ "net/http"
+ "strconv"
+
+ "github.com/gin-gonic/gin"
+ "github.com/starfleetcptn/gomft/components"
+)
+
+// HandleNotifications displays all notifications for the current user
+func (h *Handlers) HandleNotifications(c *gin.Context) {
+ userID := c.GetUint("userID")
+
+ // Get notifications for the user
+ notifications, err := h.DB.GetUserNotifications(userID, 50) // Get more for the full page
+ if err != nil {
+ c.String(http.StatusInternalServerError, "Failed to load notifications")
+ return
+ }
+
+ // Get unread count
+ unreadCount, err := h.DB.GetUnreadNotificationCount(userID)
+ if err != nil {
+ c.String(http.StatusInternalServerError, "Failed to load notification count")
+ return
+ }
+
+ data := components.NotificationsData{
+ Notifications: notifications,
+ UnreadCount: unreadCount,
+ }
+
+ // Render the notifications page
+ components.NotificationsPage(c.Request.Context(), data).Render(c, c.Writer)
+}
+
+// HandleLoadNotifications loads the notifications dropdown content
+func (h *Handlers) HandleLoadNotifications(c *gin.Context) {
+ userID := c.GetUint("userID")
+
+ // Get 10 most recent notifications
+ notifications, err := h.DB.GetUserNotifications(userID, 10)
+ if err != nil {
+ c.String(http.StatusInternalServerError, "Failed to load notifications")
+ return
+ }
+
+ // Get unread count
+ unreadCount, err := h.DB.GetUnreadNotificationCount(userID)
+ if err != nil {
+ c.String(http.StatusInternalServerError, "Failed to load notification count")
+ return
+ }
+
+ data := components.NotificationsData{
+ Notifications: notifications,
+ UnreadCount: unreadCount,
+ }
+
+ // Render just the dropdown content
+ components.NotificationDropdown(data).Render(c, c.Writer)
+}
+
+// HandleNotificationCount returns the notification count badge
+func (h *Handlers) HandleNotificationCount(c *gin.Context) {
+ userID := c.GetUint("userID")
+
+ // Get unread count
+ unreadCount, err := h.DB.GetUnreadNotificationCount(userID)
+
+ if err != nil {
+ c.String(http.StatusInternalServerError, "Failed to load notification count")
+ return
+ }
+
+ // Render just the count badge
+ components.NotificationCount(unreadCount).Render(c, c.Writer)
+}
+
+// HandleMarkNotificationAsRead marks a single notification as read
+func (h *Handlers) HandleMarkNotificationAsRead(c *gin.Context) {
+ userID := c.GetUint("userID")
+
+ // Get notification ID from path
+ idParam := c.Param("id")
+ id, err := strconv.ParseUint(idParam, 10, 64)
+ if err != nil {
+ c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid notification ID"})
+ return
+ }
+
+ // Mark as read
+ if err := h.DB.MarkNotificationAsRead(uint(id)); err != nil {
+ c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to mark notification as read"})
+ return
+ }
+
+ // Return updated count
+ unreadCount, err := h.DB.GetUnreadNotificationCount(userID)
+ if err != nil {
+ c.String(http.StatusInternalServerError, "Failed to load notification count")
+ return
+ }
+
+ // Return just the updated count badge
+ components.NotificationCount(unreadCount).Render(c, c.Writer)
+}
+
+// HandleMarkAllNotificationsAsRead marks all notifications for a user as read
+func (h *Handlers) HandleMarkAllNotificationsAsRead(c *gin.Context) {
+ userID := c.GetUint("userID")
+
+ // Mark all as read
+ if err := h.DB.MarkAllNotificationsAsRead(userID); err != nil {
+ c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to mark notifications as read"})
+ return
+ }
+
+ // Return empty count (no more unread notifications)
+ components.NotificationCount(0).Render(c, c.Writer)
+}
diff --git a/internal/web/handlers/routes.go b/internal/web/handlers/routes.go
index 6e6d87b..efd3839 100644
--- a/internal/web/handlers/routes.go
+++ b/internal/web/handlers/routes.go
@@ -35,6 +35,13 @@ func (h *Handlers) RegisterRoutes(router *gin.Engine) {
authorized.GET("/profile/2fa/backup-codes", h.Handle2FABackupCodes)
authorized.POST("/profile/2fa/regenerate-codes", h.Handle2FARegenerateCodes)
+ // Add notifications routes
+ authorized.GET("/notifications", h.HandleNotifications)
+ authorized.GET("/notifications/dropdown", h.HandleLoadNotifications)
+ authorized.GET("/notifications/count", h.HandleNotificationCount)
+ authorized.POST("/notifications/:id/read", h.HandleMarkNotificationAsRead)
+ authorized.POST("/notifications/mark-all-read", h.HandleMarkAllNotificationsAsRead)
+
{
authorized.GET("/dashboard", h.HandleDashboard)
authorized.GET("/configs", h.HandleConfigs)
@@ -44,6 +51,7 @@ func (h *Handlers) RegisterRoutes(router *gin.Engine) {
authorized.PUT("/configs/:id", h.HandleUpdateConfig)
authorized.POST("/configs/:id", h.HandleUpdateConfig)
authorized.DELETE("/configs/:id", h.HandleDeleteConfig)
+ authorized.POST("/configs/:id/duplicate", h.HandleDuplicateConfig)
// Path validation endpoint
authorized.GET("/check-path", h.HandleCheckPath)
@@ -60,6 +68,7 @@ func (h *Handlers) RegisterRoutes(router *gin.Engine) {
authorized.PUT("/jobs/:id", h.HandleUpdateJob)
authorized.POST("/jobs/:id", h.HandleUpdateJob)
authorized.DELETE("/jobs/:id", h.HandleDeleteJob)
+ authorized.POST("/jobs/:id/duplicate", h.HandleDuplicateJob)
authorized.POST("/jobs/:id/run", h.HandleRunJob)
authorized.GET("/history", h.HandleHistory)
authorized.GET("/job-runs/:id", h.HandleJobRunDetails)
@@ -132,6 +141,7 @@ func (h *Handlers) RegisterRoutes(router *gin.Engine) {
settingsGroup.GET("", h.HandleSettings)
settingsGroup.POST("/notifications", h.HandleCreateNotificationService)
settingsGroup.DELETE("/notifications/:id", h.HandleDeleteNotificationService)
+ settingsGroup.POST("/notifications/test", h.HandleTestNotification)
settingsGroup.POST("/general", h.HandleSettings) // Placeholder for future implementation
settingsGroup.POST("/security", h.HandleSettings) // Placeholder for future implementation
}
diff --git a/internal/web/handlers/settings_handlers.go b/internal/web/handlers/settings_handlers.go
index 95cc553..7f8a47a 100644
--- a/internal/web/handlers/settings_handlers.go
+++ b/internal/web/handlers/settings_handlers.go
@@ -1,56 +1,24 @@
package handlers
import (
+ "bytes"
"encoding/json"
+ "fmt"
+ "io"
"log"
"net/http"
"strconv"
+ "strings"
"time"
+ "crypto/hmac"
+ "crypto/sha256"
+
"github.com/gin-gonic/gin"
"github.com/starfleetcptn/gomft/components"
"github.com/starfleetcptn/gomft/internal/db"
)
-// NotificationService represents a notification service configuration
-type NotificationService struct {
- ID uint `json:"id" gorm:"primaryKey"`
- Name string `json:"name" gorm:"not null"`
- Type string `json:"type" gorm:"not null"` // email, slack, webhook
- IsEnabled bool `json:"is_enabled" gorm:"default:true"`
- Config map[string]string `json:"config" gorm:"-"`
- ConfigJSON string `json:"-" gorm:"column:config"`
- Description string `json:"description"`
- EventTriggers string `json:"event_triggers" gorm:"column:event_triggers;default:'[]'"`
- PayloadTemplate string `json:"payload_template" gorm:"column:payload_template"`
- SecretKey string `json:"secret_key" gorm:"column:secret_key"`
- RetryPolicy string `json:"retry_policy" gorm:"column:retry_policy;default:'simple'"`
- LastUsed time.Time `json:"last_used" gorm:"column:last_used"`
- SuccessCount int `json:"success_count" gorm:"column:success_count;default:0"`
- FailureCount int `json:"failure_count" gorm:"column:failure_count;default:0"`
- CreatedBy uint `json:"created_by"`
- CreatedAt time.Time `json:"created_at"`
- UpdatedAt time.Time `json:"updated_at"`
-}
-
-// BeforeSave converts Config map to JSON string for storage
-func (n *NotificationService) BeforeSave() error {
- configJSON, err := json.Marshal(n.Config)
- if err != nil {
- return err
- }
- n.ConfigJSON = string(configJSON)
- return nil
-}
-
-// AfterFind converts JSON string back to Config map
-func (n *NotificationService) AfterFind() error {
- if n.ConfigJSON != "" {
- return json.Unmarshal([]byte(n.ConfigJSON), &n.Config)
- }
- return nil
-}
-
// HandleSettings handles GET /settings
func (h *Handlers) HandleSettings(c *gin.Context) {
// Check if the user has permission to view settings
@@ -59,7 +27,7 @@ func (h *Handlers) HandleSettings(c *gin.Context) {
return
}
- var notificationServices []NotificationService
+ var notificationServices []db.NotificationService
if err := h.DB.Find(¬ificationServices).Error; err != nil {
log.Printf("Error fetching notification services: %v", err)
}
@@ -67,14 +35,6 @@ func (h *Handlers) HandleSettings(c *gin.Context) {
// Convert to components.NotificationService
var componentServices []components.NotificationService
for _, service := range notificationServices {
- // Parse event triggers from JSON string to string slice
- var eventTriggers []string
- if service.EventTriggers != "" {
- if err := json.Unmarshal([]byte(service.EventTriggers), &eventTriggers); err != nil {
- log.Printf("Error parsing event triggers: %v", err)
- }
- }
-
componentServices = append(componentServices, components.NotificationService{
ID: service.ID,
Name: service.Name,
@@ -82,7 +42,7 @@ func (h *Handlers) HandleSettings(c *gin.Context) {
IsEnabled: service.IsEnabled,
Config: service.Config,
Description: service.Description,
- EventTriggers: eventTriggers,
+ EventTriggers: service.EventTriggers,
PayloadTemplate: service.PayloadTemplate,
SecretKey: service.SecretKey,
RetryPolicy: service.RetryPolicy,
@@ -130,16 +90,17 @@ func (h *Handlers) HandleCreateNotificationService(c *gin.Context) {
config["smtp_username"] = c.PostForm("smtp_username")
config["smtp_password"] = c.PostForm("smtp_password")
config["from_email"] = c.PostForm("from_email")
- case "slack":
- config["webhook_url"] = c.PostForm("webhook_url")
- config["channel"] = c.PostForm("channel")
case "webhook":
config["webhook_url"] = c.PostForm("webhook_url")
config["method"] = c.PostForm("method")
config["headers"] = c.PostForm("headers")
// Add the new webhook fields
- // Create event triggers JSON array
+ // Create event triggers array
+ // print all event triggers
+ log.Printf("Event triggers: %v", c.PostForm("trigger_job_start"))
+ log.Printf("Event triggers: %v", c.PostForm("trigger_job_complete"))
+ log.Printf("Event triggers: %v", c.PostForm("trigger_job_error"))
eventTriggers := make([]string, 0)
if c.PostForm("trigger_job_start") == "on" {
eventTriggers = append(eventTriggers, "job_start")
@@ -151,21 +112,14 @@ func (h *Handlers) HandleCreateNotificationService(c *gin.Context) {
eventTriggers = append(eventTriggers, "job_error")
}
- // Marshal the event triggers to JSON
- eventTriggersJSON, err := json.Marshal(eventTriggers)
- if err != nil {
- h.handleSettingsWithError(c, "Failed to process event triggers: "+err.Error())
- return
- }
-
// Create new notification service with additional fields
- service := NotificationService{
+ service := db.NotificationService{
Name: name,
Type: serviceType,
IsEnabled: isEnabled,
Config: config,
Description: description,
- EventTriggers: string(eventTriggersJSON),
+ EventTriggers: eventTriggers,
PayloadTemplate: c.PostForm("payload_template"),
SecretKey: c.PostForm("secret_key"),
RetryPolicy: c.PostForm("retry_policy"),
@@ -211,7 +165,7 @@ func (h *Handlers) HandleCreateNotificationService(c *gin.Context) {
}
// Create new notification service
- service := NotificationService{
+ service := db.NotificationService{
Name: name,
Type: serviceType,
IsEnabled: isEnabled,
@@ -267,14 +221,14 @@ func (h *Handlers) HandleDeleteNotificationService(c *gin.Context) {
}
// Find service to delete (for audit log)
- var service NotificationService
+ var service db.NotificationService
if err := h.DB.First(&service, serviceID).Error; err != nil {
h.handleSettingsWithError(c, "Notification service not found.")
return
}
// Delete the service
- if err := h.DB.Delete(&NotificationService{}, serviceID).Error; err != nil {
+ if err := h.DB.Delete(&db.NotificationService{}, serviceID).Error; err != nil {
log.Printf("Error deleting notification service: %v", err)
h.handleSettingsWithError(c, "Failed to delete notification service: "+err.Error())
return
@@ -304,6 +258,224 @@ func (h *Handlers) HandleDeleteNotificationService(c *gin.Context) {
h.handleSettingsWithSuccess(c, "Notification service deleted successfully.")
}
+// HandleTestNotification handles POST /settings/notifications/test
+// This endpoint tests a notification configuration without saving it
+func (h *Handlers) HandleTestNotification(c *gin.Context) {
+ // Check if the user has permission to manage settings
+ if !h.checkPermission(c, "system.settings") {
+ c.JSON(http.StatusForbidden, gin.H{
+ "success": false,
+ "message": "You don't have permission to test notifications",
+ })
+ return
+ }
+
+ // Parse form data to create a test notification service
+ name := c.PostForm("name")
+ serviceType := c.PostForm("type")
+
+ // Validate required fields
+ if name == "" || serviceType == "" {
+ c.JSON(http.StatusBadRequest, gin.H{
+ "success": false,
+ "message": "Name and type are required fields",
+ })
+ return
+ }
+
+ // Create config map based on service type
+ config := make(map[string]string)
+
+ switch serviceType {
+ case "email":
+ config["smtp_host"] = c.PostForm("smtp_host")
+ config["smtp_port"] = c.PostForm("smtp_port")
+ config["smtp_username"] = c.PostForm("smtp_username")
+ config["smtp_password"] = c.PostForm("smtp_password")
+ config["from_email"] = c.PostForm("from_email")
+
+ // Basic validation
+ if config["smtp_host"] == "" || config["smtp_port"] == "" {
+ c.JSON(http.StatusBadRequest, gin.H{
+ "success": false,
+ "message": "SMTP host and port are required for email notifications",
+ })
+ return
+ }
+
+ case "webhook":
+ config["webhook_url"] = c.PostForm("webhook_url")
+ config["method"] = c.PostForm("method")
+ config["headers"] = c.PostForm("headers")
+
+ // Basic validation
+ if config["webhook_url"] == "" {
+ c.JSON(http.StatusBadRequest, gin.H{
+ "success": false,
+ "message": "Webhook URL is required for webhook notifications",
+ })
+ return
+ }
+
+ // Get additional webhook fields
+ payloadTemplate := c.PostForm("payload_template")
+ secretKey := c.PostForm("secret_key")
+
+ // Create and format sample payload
+ samplePayload := generateSamplePayload(payloadTemplate)
+
+ // Send test webhook
+ err := sendTestWebhook(config, samplePayload, secretKey)
+ if err != nil {
+ c.JSON(http.StatusInternalServerError, gin.H{
+ "success": false,
+ "message": "Failed to send test webhook: " + err.Error(),
+ })
+ return
+ }
+
+ c.JSON(http.StatusOK, gin.H{
+ "success": true,
+ "message": "Test webhook sent successfully",
+ })
+ return
+
+ default:
+ c.JSON(http.StatusBadRequest, gin.H{
+ "success": false,
+ "message": "Invalid notification service type",
+ })
+ return
+ }
+
+ // For email, simulate a successful test for now
+ // In a real implementation, you would send an actual test notification
+ c.JSON(http.StatusOK, gin.H{
+ "success": true,
+ "message": fmt.Sprintf("Simulated %s notification test successful", serviceType),
+ })
+}
+
+// generateSamplePayload creates a sample payload for testing
+func generateSamplePayload(template string) string {
+ // If no template provided, use a default sample
+ if template == "" {
+ return `{
+ "event": "job_complete",
+ "job": {
+ "id": "sample-job-123",
+ "name": "Test Job",
+ "status": "completed",
+ "message": "This is a test notification",
+ "started_at": "` + time.Now().Add(-5*time.Minute).Format(time.RFC3339) + `",
+ "completed_at": "` + time.Now().Format(time.RFC3339) + `",
+ "duration_seconds": 300,
+ "config_id": "config-456",
+ "config_name": "Test Config",
+ "transfer_bytes": 1024,
+ "file_count": 5
+ },
+ "instance": {
+ "id": "gomft-instance-1",
+ "name": "GoMFT Test Instance",
+ "version": "1.0.0",
+ "environment": "testing"
+ },
+ "timestamp": "` + time.Now().Format(time.RFC3339) + `",
+ "notification_id": "test-notification"
+ }`
+ }
+
+ // Replace placeholders in the template with sample values
+ samplePayload := template
+ // Replace common placeholders
+ replacements := map[string]string{
+ "{{job.id}}": "sample-job-123",
+ "{{job.name}}": "Test Job",
+ "{{job.status}}": "completed",
+ "{{job.message}}": "This is a test notification",
+ "{{job.event}}": "job_complete",
+ "{{job.started_at}}": time.Now().Add(-5 * time.Minute).Format(time.RFC3339),
+ "{{job.completed_at}}": time.Now().Format(time.RFC3339),
+ "{{job.duration_seconds}}": "300",
+ "{{job.config_id}}": "config-456",
+ "{{job.config_name}}": "Test Config",
+ "{{job.transfer_bytes}}": "1024",
+ "{{job.file_count}}": "5",
+ "{{instance.id}}": "gomft-instance-1",
+ "{{instance.name}}": "GoMFT Test Instance",
+ "{{instance.version}}": "1.0.0",
+ "{{instance.environment}}": "testing",
+ "{{timestamp}}": time.Now().Format(time.RFC3339),
+ "{{notification.id}}": "test-notification",
+ }
+
+ for placeholder, value := range replacements {
+ samplePayload = strings.Replace(samplePayload, placeholder, value, -1)
+ }
+
+ return samplePayload
+}
+
+// sendTestWebhook sends a test webhook to the specified URL
+func sendTestWebhook(config map[string]string, payload string, secretKey string) error {
+ webhookURL := config["webhook_url"]
+ method := config["method"]
+ if method == "" {
+ method = "POST"
+ }
+
+ // Create the request
+ req, err := http.NewRequest(method, webhookURL, bytes.NewBufferString(payload))
+ if err != nil {
+ return fmt.Errorf("error creating request: %v", err)
+ }
+
+ // Set default Content-Type if not specified
+ req.Header.Set("Content-Type", "application/json")
+
+ // Parse and set custom headers
+ if config["headers"] != "" {
+ var headers map[string]string
+ if err := json.Unmarshal([]byte(config["headers"]), &headers); err == nil {
+ for key, value := range headers {
+ req.Header.Set(key, value)
+ }
+ }
+ }
+
+ // Add signature if secret key is provided
+ if secretKey != "" {
+ signature := calculateSignature(payload, secretKey)
+ req.Header.Set("X-GoMFT-Signature", signature)
+ }
+
+ // Send the request
+ client := &http.Client{
+ Timeout: 10 * time.Second,
+ }
+ resp, err := client.Do(req)
+ if err != nil {
+ return fmt.Errorf("error sending webhook: %v", err)
+ }
+ defer resp.Body.Close()
+
+ // Check the response
+ if resp.StatusCode >= 400 {
+ body, _ := io.ReadAll(resp.Body)
+ return fmt.Errorf("webhook returned error %d: %s", resp.StatusCode, string(body))
+ }
+
+ return nil
+}
+
+// calculateSignature generates an HMAC signature for webhook payloads
+func calculateSignature(payload string, secretKey string) string {
+ h := hmac.New(sha256.New, []byte(secretKey))
+ h.Write([]byte(payload))
+ return fmt.Sprintf("sha256=%x", h.Sum(nil))
+}
+
// Helper function to check if user has a specific permission
func (h *Handlers) checkPermission(c *gin.Context, permission string) bool {
// If user is admin, they have all permissions
@@ -332,7 +504,7 @@ func (h *Handlers) checkPermission(c *gin.Context, permission string) bool {
// handleSettingsWithError renders the settings page with an error message
func (h *Handlers) handleSettingsWithError(c *gin.Context, errorMessage string) {
- var notificationServices []NotificationService
+ var notificationServices []db.NotificationService
if err := h.DB.Find(¬ificationServices).Error; err != nil {
log.Printf("Error fetching notification services: %v", err)
}
@@ -340,14 +512,6 @@ func (h *Handlers) handleSettingsWithError(c *gin.Context, errorMessage string)
// Convert to components.NotificationService
var componentServices []components.NotificationService
for _, service := range notificationServices {
- // Parse event triggers from JSON string to string slice
- var eventTriggers []string
- if service.EventTriggers != "" {
- if err := json.Unmarshal([]byte(service.EventTriggers), &eventTriggers); err != nil {
- log.Printf("Error parsing event triggers: %v", err)
- }
- }
-
componentServices = append(componentServices, components.NotificationService{
ID: service.ID,
Name: service.Name,
@@ -355,7 +519,7 @@ func (h *Handlers) handleSettingsWithError(c *gin.Context, errorMessage string)
IsEnabled: service.IsEnabled,
Config: service.Config,
Description: service.Description,
- EventTriggers: eventTriggers,
+ EventTriggers: service.EventTriggers,
PayloadTemplate: service.PayloadTemplate,
SecretKey: service.SecretKey,
RetryPolicy: service.RetryPolicy,
@@ -375,7 +539,7 @@ func (h *Handlers) handleSettingsWithError(c *gin.Context, errorMessage string)
// handleSettingsWithSuccess renders the settings page with a success message
func (h *Handlers) handleSettingsWithSuccess(c *gin.Context, successMessage string) {
- var notificationServices []NotificationService
+ var notificationServices []db.NotificationService
if err := h.DB.Find(¬ificationServices).Error; err != nil {
log.Printf("Error fetching notification services: %v", err)
}
@@ -383,14 +547,6 @@ func (h *Handlers) handleSettingsWithSuccess(c *gin.Context, successMessage stri
// Convert to components.NotificationService
var componentServices []components.NotificationService
for _, service := range notificationServices {
- // Parse event triggers from JSON string to string slice
- var eventTriggers []string
- if service.EventTriggers != "" {
- if err := json.Unmarshal([]byte(service.EventTriggers), &eventTriggers); err != nil {
- log.Printf("Error parsing event triggers: %v", err)
- }
- }
-
componentServices = append(componentServices, components.NotificationService{
ID: service.ID,
Name: service.Name,
@@ -398,7 +554,7 @@ func (h *Handlers) handleSettingsWithSuccess(c *gin.Context, successMessage stri
IsEnabled: service.IsEnabled,
Config: service.Config,
Description: service.Description,
- EventTriggers: eventTriggers,
+ EventTriggers: service.EventTriggers,
PayloadTemplate: service.PayloadTemplate,
SecretKey: service.SecretKey,
RetryPolicy: service.RetryPolicy,