diff --git a/server/service-operation/data-retention/handler.go b/server/service-operation/data-retention/handler.go new file mode 100644 index 0000000..514db3d --- /dev/null +++ b/server/service-operation/data-retention/handler.go @@ -0,0 +1,59 @@ +package dataretention + +import ( + "encoding/json" + "net/http" + "service-operation/pocketbase" +) + +// Handler handles HTTP requests for data retention operations +type Handler struct { + service *Service +} + +// NewHandler creates a new data retention handler +func NewHandler(pbClient *pocketbase.PocketBaseClient) *Handler { + return &Handler{ + service: NewService(pbClient), + } +} + +// GetSettings handles GET requests for data retention settings +func (h *Handler) GetSettings(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + settings, err := h.service.GetDataSettings() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(settings) +} + +// TriggerCleanup handles POST requests to trigger manual cleanup +func (h *Handler) TriggerCleanup(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + summary, err := h.service.CleanupOldRecords() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(summary) +} + +// SetupRoutes sets up the HTTP routes for data retention +func (h *Handler) SetupRoutes(mux *http.ServeMux) { + mux.HandleFunc("/data-retention/settings", h.GetSettings) + mux.HandleFunc("/data-retention/cleanup", h.TriggerCleanup) +} \ No newline at end of file diff --git a/server/service-operation/data-retention/scheduler.go b/server/service-operation/data-retention/scheduler.go new file mode 100644 index 0000000..a064cee --- /dev/null +++ b/server/service-operation/data-retention/scheduler.go @@ -0,0 +1,63 @@ +package dataretention + +import ( + "service-operation/pocketbase" + "time" +) + +// Scheduler handles periodic data retention cleanup +type Scheduler struct { + service *Service + interval time.Duration + stopChan chan bool +} + +// NewScheduler creates a new cleanup scheduler +func NewScheduler(pbClient *pocketbase.PocketBaseClient, interval time.Duration) *Scheduler { + return &Scheduler{ + service: NewService(pbClient), + interval: interval, + stopChan: make(chan bool), + } +} + +// Start begins the periodic cleanup process +func (s *Scheduler) Start() { + // log.Printf("Starting data retention scheduler with interval: %v", s.interval) + + ticker := time.NewTicker(s.interval) + defer ticker.Stop() + + // Run initial cleanup + go func() { + if err := s.service.ScheduleCleanup(); err != nil { + // log.Printf("Initial cleanup failed: %v", err) + } + }() + + for { + select { + case <-ticker.C: + go func() { + if err := s.service.ScheduleCleanup(); err != nil { + // log.Printf("Scheduled cleanup failed: %v", err) + } + }() + case <-s.stopChan: + // log.Println("Data retention scheduler stopped") + return + } + } +} + +// Stop stops the scheduler +func (s *Scheduler) Stop() { + // log.Println("Stopping data retention scheduler...") + s.stopChan <- true +} + +// RunOnce executes cleanup immediately (for manual triggers) +func (s *Scheduler) RunOnce() (*CleanupSummary, error) { + // log.Println("Running manual data retention cleanup...") + return s.service.CleanupOldRecords() +} \ No newline at end of file diff --git a/server/service-operation/data-retention/service.go b/server/service-operation/data-retention/service.go new file mode 100644 index 0000000..75e224c --- /dev/null +++ b/server/service-operation/data-retention/service.go @@ -0,0 +1,398 @@ +package dataretention + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/url" + "service-operation/pocketbase" + "sync" + "time" +) + +// Service handles data retention operations +type Service struct { + pbClient *pocketbase.PocketBaseClient +} + +// NewService creates a new data retention service +func NewService(pbClient *pocketbase.PocketBaseClient) *Service { + return &Service{ + pbClient: pbClient, + } +} + +// GetDataSettings fetches the current data retention settings +func (s *Service) GetDataSettings() (*DataSettings, error) { + resp, err := s.pbClient.GetHTTPClient().Get( + fmt.Sprintf("%s/api/collections/data_settings/records", s.pbClient.GetBaseURL()), + ) + if err != nil { + return nil, fmt.Errorf("failed to fetch data settings: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("failed to fetch data settings, status: %d", resp.StatusCode) + } + + var response DataSettingsResponse + if err := json.NewDecoder(resp.Body).Decode(&response); err != nil { + return nil, fmt.Errorf("failed to decode data settings response: %w", err) + } + + if len(response.Items) == 0 { + return nil, fmt.Errorf("no data settings found") + } + + return &response.Items[0], nil +} + +// CleanupOldRecords performs cleanup of old records based on retention settings +func (s *Service) CleanupOldRecords() (*CleanupSummary, error) { + startTime := time.Now() + + settings, err := s.GetDataSettings() + if err != nil { + return nil, fmt.Errorf("failed to get data settings: %w", err) + } + + // Check if cleanup should run (only once per day) + if !s.shouldRunCleanup(settings) { + // log.Println("Skipping cleanup - less than 24 hours since last run") + return &CleanupSummary{ + StartTime: startTime, + EndTime: time.Now(), + Duration: 0, + Results: make([]CleanupResult, 0), + }, nil + } + + summary := &CleanupSummary{ + StartTime: startTime, + Results: make([]CleanupResult, 0), + } + + // Define collections and their corresponding retention days + uptimeCollections := []string{"dns_data", "ping_data", "tcp_data", "uptime_data", "services_metrics"} + serverCollections := []string{"server_metrics", "docker_metrics"} + + // Process collections in parallel for better performance + var wg sync.WaitGroup + resultsChan := make(chan CleanupResult, len(uptimeCollections)+len(serverCollections)) + + // Cleanup uptime-related collections in parallel + if settings.UptimeRetentionDays > 0 { + cutoffDate := time.Now().AddDate(0, 0, -settings.UptimeRetentionDays) + // log.Printf("Cleaning uptime collections with %d days retention (cutoff: %s)", + // settings.UptimeRetentionDays, cutoffDate.Format("2006-01-02 15:04:05")) + for _, collection := range uptimeCollections { + wg.Add(1) + go func(col string) { + defer wg.Done() + result := s.cleanupCollection(col, cutoffDate) + resultsChan <- result + }(collection) + } + } else { + // log.Println("Skipping uptime collections cleanup - retention days is 0") + } + + // Cleanup server-related collections in parallel + if settings.ServerRetentionDays > 0 { + cutoffDate := time.Now().AddDate(0, 0, -settings.ServerRetentionDays) + // log.Printf("Cleaning server collections with %d days retention (cutoff: %s)", + // settings.ServerRetentionDays, cutoffDate.Format("2006-01-02 15:04:05")) + for _, collection := range serverCollections { + wg.Add(1) + go func(col string) { + defer wg.Done() + result := s.cleanupCollection(col, cutoffDate) + resultsChan <- result + }(collection) + } + } else { + // log.Println("Skipping server collections cleanup - retention days is 0") + } + + // Wait for all collections to complete and collect results + go func() { + wg.Wait() + close(resultsChan) + }() + + // Collect all results + for result := range resultsChan { + summary.Results = append(summary.Results, result) + summary.TotalDeleted += result.RecordsDeleted + } + + summary.EndTime = time.Now() + summary.Duration = summary.EndTime.Sub(summary.StartTime) + + // Update last cleanup time after successful completion + if err := s.updateLastCleanupTime(); err != nil { + // log.Printf("Warning: Failed to update last cleanup time: %v", err) + } + + // log.Printf("Data retention cleanup completed. Total records deleted: %d, Duration: %v", + // summary.TotalDeleted, summary.Duration) + + return summary, nil +} + +// cleanupCollection deletes records older than the cutoff date from a specific collection +func (s *Service) cleanupCollection(collection string, cutoffDate time.Time) CleanupResult { + result := CleanupResult{ + Collection: collection, + } + + // Format cutoff date for PocketBase filter (RFC3339 format) + cutoffStr := cutoffDate.UTC().Format("2006-01-02T15:04:05.000Z") + + // Get records to delete (balanced batch size for performance vs rate limits) + batchSize := 50 + totalDeleted := 0 + + for { + // Always fetch page 1 since we're deleting records (pagination shifts after deletes) + filter := fmt.Sprintf("created < '%s'", cutoffStr) + encodedFilter := url.QueryEscape(filter) + requestURL := fmt.Sprintf("%s/api/collections/%s/records?page=1&perPage=%d&filter=%s", + s.pbClient.GetBaseURL(), collection, batchSize, encodedFilter) + + resp, err := s.pbClient.GetHTTPClient().Get(requestURL) + if err != nil { + result.Error = fmt.Sprintf("failed to fetch records: %v", err) + return result + } + defer resp.Body.Close() + + // Check if response is successful + if resp.StatusCode != http.StatusOK { + result.Error = fmt.Sprintf("API request failed with status %d for collection %s", resp.StatusCode, collection) + return result + } + + var response struct { + Items []struct { + ID string `json:"id"` + } `json:"items"` + TotalItems int `json:"totalItems"` + } + + if err := json.NewDecoder(resp.Body).Decode(&response); err != nil { + result.Error = fmt.Sprintf("failed to decode response for collection %s: %v", collection, err) + return result + } + + if len(response.Items) == 0 { + // log.Printf("No more old records found in collection %s", collection) + break // No more records to delete + } + + // log.Printf("Found %d old records in collection %s (batch), total items in DB: %d", + // len(response.Items), collection, response.TotalItems) + + // Delete records with conservative rate limiting to avoid 429 errors + semaphore := make(chan struct{}, 3) // Further reduced to 3 concurrent deletes + var deleteWg sync.WaitGroup + var mu sync.Mutex + batchDeleted := 0 + rateLimitHit := false + + for _, record := range response.Items { + deleteWg.Add(1) + go func(recordID string) { + defer deleteWg.Done() + semaphore <- struct{}{} // Acquire + defer func() { <-semaphore }() // Release + + deleted := false + retryCount := 0 + maxRetries := 5 + + for !deleted && retryCount < maxRetries { + if err := s.deleteRecord(collection, recordID); err != nil { + if retryCount < maxRetries-1 { + // Progressive backoff with longer delays for rate limits + baseWait := time.Duration(1<= 2 { + // Extra penalty for persistent rate limits + baseWait += time.Duration(retryCount) * 2 * time.Second + } + + mu.Lock() + rateLimitHit = true + mu.Unlock() + + time.Sleep(baseWait) + retryCount++ + continue + } else { + // log.Printf("Failed to delete record %s from %s after %d retries: %v", + // recordID, collection, maxRetries, err) + break + } + } else { + deleted = true + mu.Lock() + batchDeleted++ + mu.Unlock() + } + } + + // Increased delay between deletions with adaptive timing + mu.Lock() + currentRateLimitHit := rateLimitHit + mu.Unlock() + + if currentRateLimitHit { + time.Sleep(150 * time.Millisecond) // Longer delay if rate limits detected + } else { + time.Sleep(75 * time.Millisecond) // Standard delay + } + }(record.ID) + } + + deleteWg.Wait() + totalDeleted += batchDeleted + + // If we got fewer records than batch size, we're done + if len(response.Items) < batchSize { + // log.Printf("Finished processing collection %s - no more old records", collection) + break + } + + // More conservative delay between batches with adaptive timing + if len(response.Items) > 0 { + mu.Lock() + currentRateLimitHit := rateLimitHit + mu.Unlock() + + baseDelay := 300 * time.Millisecond + if currentRateLimitHit { + baseDelay = 1 * time.Second // Much longer delay if rate limits were hit + // log.Printf("Processed batch for collection %s (%d deleted), rate limits detected - waiting longer before next batch...", collection, batchDeleted) + } else { + // log.Printf("Processed batch for collection %s (%d deleted), processing next batch...", collection, batchDeleted) + } + + time.Sleep(baseDelay) + } + } + + result.RecordsDeleted = totalDeleted + + if totalDeleted > 0 { + // log.Printf("Deleted %d records from collection %s", totalDeleted, collection) + } + + return result +} + +// deleteRecord deletes a single record from a collection +func (s *Service) deleteRecord(collection, recordID string) error { + requestURL := fmt.Sprintf("%s/api/collections/%s/records/%s", s.pbClient.GetBaseURL(), collection, recordID) + + req, err := http.NewRequest("DELETE", requestURL, nil) + if err != nil { + return err + } + + resp, err := s.pbClient.GetHTTPClient().Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK { + return fmt.Errorf("failed to delete record, status: %d", resp.StatusCode) + } + + return nil +} + +// ScheduleCleanup runs cleanup based on schedule (can be called periodically) +func (s *Service) ScheduleCleanup() error { + // log.Println("Starting scheduled data retention cleanup...") + + _, err := s.CleanupOldRecords() + if err != nil { + // log.Printf("Scheduled cleanup failed: %v", err) + return err + } + + // log.Printf("Scheduled cleanup completed successfully. Summary: %+v", summary) + return nil +} + +// shouldRunCleanup checks if cleanup should run based on last cleanup time (once per day) +func (s *Service) shouldRunCleanup(settings *DataSettings) bool { + if settings.LastCleanup == "" { + // log.Println("No previous cleanup found, proceeding with cleanup...") + return true + } + + lastCleanup, err := time.Parse(time.RFC3339, settings.LastCleanup) + if err != nil { + // log.Printf("Failed to parse last cleanup time '%s', proceeding with cleanup: %v", settings.LastCleanup, err) + return true + } + + timeSinceLastCleanup := time.Since(lastCleanup) + minInterval := 24 * time.Hour + + if timeSinceLastCleanup < minInterval { + // log.Printf("Last cleanup was %v ago (less than 24h), skipping cleanup", timeSinceLastCleanup) + return false + } + + // log.Printf("Last cleanup was %v ago, proceeding with cleanup", timeSinceLastCleanup) + return true +} + +// updateLastCleanupTime updates the last cleanup timestamp in data settings +func (s *Service) updateLastCleanupTime() error { + settings, err := s.GetDataSettings() + if err != nil { + return fmt.Errorf("failed to get data settings: %w", err) + } + + // Update the last cleanup time + now := time.Now().UTC().Format(time.RFC3339) + + // Create the update payload + updateData := map[string]interface{}{ + "last_cleanup": now, + } + + jsonData, err := json.Marshal(updateData) + if err != nil { + return fmt.Errorf("failed to marshal update data: %w", err) + } + + // Update the record via PocketBase API + requestURL := fmt.Sprintf("%s/api/collections/data_settings/records/%s", s.pbClient.GetBaseURL(), settings.ID) + + req, err := http.NewRequest("PATCH", requestURL, bytes.NewBuffer(jsonData)) + if err != nil { + return fmt.Errorf("failed to create update request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + + resp, err := s.pbClient.GetHTTPClient().Do(req) + if err != nil { + return fmt.Errorf("failed to update last cleanup time: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("failed to update last cleanup time, status: %d", resp.StatusCode) + } + + // log.Printf("Updated last cleanup time to: %s", now) + return nil +} \ No newline at end of file diff --git a/server/service-operation/data-retention/types.go b/server/service-operation/data-retention/types.go new file mode 100644 index 0000000..7ebdb1c --- /dev/null +++ b/server/service-operation/data-retention/types.go @@ -0,0 +1,42 @@ +package dataretention + +import "time" + +// DataSettings represents the data retention configuration from PocketBase +type DataSettings struct { + ID string `json:"id"` + CollectionId string `json:"collectionId"` + CollectionName string `json:"collectionName"` + RetentionDays int `json:"retention_days"` + ServerRetentionDays int `json:"server_retention_days"` + UptimeRetentionDays int `json:"uptime_retention_days"` + LastCleanup string `json:"last_cleanup"` + Backup string `json:"backup"` + Created string `json:"created"` + Updated string `json:"updated"` +} + +// DataSettingsResponse represents the response from PocketBase +type DataSettingsResponse struct { + Items []DataSettings `json:"items"` + Page int `json:"page"` + PerPage int `json:"perPage"` + TotalItems int `json:"totalItems"` + TotalPages int `json:"totalPages"` +} + +// CleanupResult represents the result of a cleanup operation +type CleanupResult struct { + Collection string `json:"collection"` + RecordsDeleted int `json:"records_deleted"` + Error string `json:"error,omitempty"` +} + +// CleanupSummary represents the overall cleanup summary +type CleanupSummary struct { + TotalDeleted int `json:"total_deleted"` + Results []CleanupResult `json:"results"` + StartTime time.Time `json:"start_time"` + EndTime time.Time `json:"end_time"` + Duration time.Duration `json:"duration"` +} \ No newline at end of file diff --git a/server/service-operation/main.go b/server/service-operation/main.go index e337318..37999aa 100644 --- a/server/service-operation/main.go +++ b/server/service-operation/main.go @@ -6,6 +6,7 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/gorilla/mux" "service-operation/config" @@ -15,6 +16,7 @@ import ( servermonitoring "service-operation/server-monitoring" sslmonitoring "service-operation/ssl-monitoring" uptimemonitoring "service-operation/uptime-monitoring" + dataretention "service-operation/data-retention" ) func main() { @@ -35,6 +37,7 @@ func main() { var sslNotificationService *sslmonitoring.SSLMonitor var serverMonitoringService *servermonitoring.ServerMonitoringService var uptimeMonitoringService *uptimemonitoring.UptimeMonitor + var dataRetentionScheduler *dataretention.Scheduler if cfg.PocketBaseEnabled { //log.Println("🔧 Initializing PocketBase client...") @@ -80,6 +83,12 @@ func main() { uptimeMonitoringService = uptimemonitoring.NewUptimeMonitor(pbClient) go uptimeMonitoringService.Start() //log.Println("✅ Uptime monitoring started with notification support") + + // Initialize and start data retention scheduler + //log.Println("🔧 Initializing data retention scheduler...") + dataRetentionScheduler = dataretention.NewScheduler(pbClient, 24*time.Hour) // Run daily + go dataRetentionScheduler.Start() + //log.Println("✅ Data retention scheduler started (daily cleanup)") } } } @@ -123,6 +132,9 @@ func main() { log.Printf("✓Uptime monitoring enabled with notification support") } log.Printf("✓Supported operations: ping, dns, tcp, http, ssl") + if dataRetentionScheduler != nil { + log.Printf("✓Data retention scheduler enabled (daily cleanup)") + } // Setup graceful shutdown @@ -154,6 +166,10 @@ func main() { log.Println("🛑 Stopping uptime monitoring...") uptimeMonitoringService.Stop() } + if dataRetentionScheduler != nil { + log.Println("🛑 Stopping data retention scheduler...") + dataRetentionScheduler.Stop() + } log.Println("✅ All services stopped gracefully") log.Println("🛑 === SERVICE OPERATION SERVER STOPPED ===")