feat: implement persistent job queue with bbolt and maintenance worker

backend/README.md

@@ -23,6 +23,11 @@
  # Else if using npm
  FRONTEND_ORIGIN_DEV="http://localhost:5173"
  CONTAINER_ORIGIN="http://localhost:8080/"
+
+ # Job Queue Configuration (Optional)
+ CLEANUP_CRON_SCHEDULE="0 0 * * *" 
+ CLEANUP_RETENTION_DAYS="7" 
+ QUEUE_DB_PATH="/app/data/queue.db" 
  ```
 
  Common pitfall: use the value
@@ -85,6 +90,28 @@
  - **Production with nginx on same server**: Not needed (loopback is trusted)
  - **Production with external load balancer**: Set to your load balancer's IP/range
 
+## Persistent Job Queue
+
+The backend includes a persistent job queue system that ensures task operations survive server restarts and provides automatic cleanup of old job logs.
+
+### Features
+
+- **Persistence**: Jobs are stored in a bbolt database and survive backend restarts
+- **Automatic Cleanup**: Old completed and failed job logs are automatically cleaned up
+- **Configurable**: Cleanup schedule and retention period can be customized
+
+### Configuration
+
+The job queue system uses the following environment variables:
+
+- `CLEANUP_CRON_SCHEDULE`: Cron schedule for cleanup job (default: "0 0 * * *" - daily at midnight)
+- `CLEANUP_RETENTION_DAYS`: Number of days to keep job logs (default: 7)
+- `QUEUE_DB_PATH`: Path to the queue database file (default: "/app/data/queue.db")
+
+### Database Location
+
+The queue database is stored at `/app/data/queue.db` inside the container, which is mounted to `./backend/data/queue.db` on the host system via Docker volume. The `QUEUE_DB_PATH` environment variable can be used to customize this location if needed.
+
 - Run the application:
 
  ```bash

backend/controllers/controllers_test.go

@@ -17,6 +17,7 @@ import (
 )
 
 func setup() *App {
+	os.Setenv("GO_ENV", "test")
 	godotenv.Load("../.env")
 
 	clientID := os.Getenv("CLIENT_ID")

backend/controllers/job_queue.go

@@ -1,7 +1,13 @@
 package controllers
 
 import (
+	"os"
 	"sync"
+	"time"
+
+	"ccsync_backend/utils"
+
+	"github.com/google/uuid"
 )
 
 type Job struct {
@@ -10,20 +16,57 @@ type Job struct {
 }
 
 type JobQueue struct {
-	jobChannel chan Job
-	wg sync.WaitGroup
+	jobChannel chan Job
+	wg sync.WaitGroup
+	persistentQueue utils.PersistentJobQueue
 }
 
 func NewJobQueue() *JobQueue {
+	dbPath := os.Getenv("QUEUE_DB_PATH")
+	if dbPath == "" {
+ dbPath = "/app/data/queue.db"
+	}
+
+	var persistentQueue utils.PersistentJobQueue
+	if os.Getenv("GO_ENV") != "test" {
+ pq, err := utils.NewBoltJobQueue(dbPath)
+ if err != nil {
+ utils.Logger.Errorf("Failed to initialize persistent queue: %v", err)
+ } else {
+ persistentQueue = pq
+ }
+	}
+
 	queue := &JobQueue{
- jobChannel: make(chan Job, 100),
+ jobChannel: make(chan Job, 100),
+ persistentQueue: persistentQueue,
+	}
+
+	if persistentQueue != nil {
+ queue.restorePendingJobs()
 	}
+
 	go queue.processJobs()
 	return queue
 }
 
 func (q *JobQueue) AddJob(job Job) {
 	q.wg.Add(1)
+
+	if q.persistentQueue != nil {
+ persistentJob := &utils.PersistentJob{
+ ID: uuid.New().String(),
+ Name: job.Name,
+ State: utils.JobStatePending,
+ CreatedAt: time.Now(),
+ UpdatedAt: time.Now(),
+ }
+
+ if err := q.persistentQueue.AddJob(persistentJob); err != nil {
+ utils.Logger.Errorf("Failed to persist job: %v", err)
+ }
+	}
+
 	q.jobChannel <- job
 
 	// notify job queued
@@ -56,3 +99,32 @@ func (q *JobQueue) processJobs() {
  q.wg.Done()
 	}
 }
+
+func (q *JobQueue) restorePendingJobs() {
+	if q.persistentQueue == nil {
+ return
+	}
+
+	pendingJobs, err := q.persistentQueue.GetPendingJobs()
+	if err != nil {
+ utils.Logger.Errorf("Failed to restore pending jobs: %v", err)
+ return
+	}
+
+	utils.Logger.Infof("Restoring %d pending jobs", len(pendingJobs))
+	for _, persistentJob := range pendingJobs {
+ job := Job{
+ Name: persistentJob.Name,
+ Execute: func() error {
+ return nil
+ },
+ }
+ q.wg.Add(1)
+ go func(j Job) {
+ q.jobChannel <- j
+ }(job)
+	}
+}
+func (q *JobQueue) GetPersistentQueue() utils.PersistentJobQueue {
+	return q.persistentQueue
+}

backend/go.mod

@@ -5,8 +5,10 @@ go 1.19
 require (
 	github.com/charmbracelet/log v0.4.2
 	github.com/gorilla/sessions v1.2.2
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/swaggo/http-swagger v1.3.4
 	github.com/swaggo/swag v1.16.3
+	go.etcd.io/bbolt v1.3.7
 	golang.org/x/oauth2 v0.20.0
 )
 

backend/go.sum

@@ -70,6 +70,8 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
 github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@@ -83,6 +85,8 @@ github.com/swaggo/swag v1.16.3 h1:PnCYjPCah8FK4I26l2F/KQ4yz3sILcVUN3cTlBFA9Pg=
 github.com/swaggo/swag v1.16.3/go.mod h1:DImHIuOFXKpMFAQjcC7FG4m3Dg4+QuUgUzJmKjI/gRk=
 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no=
 github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM=
+go.etcd.io/bbolt v1.3.7 h1:j+zJOnnEjF/kyHlDDgGnVL/AIqIJPq8UoB2GSNfkUfQ=
+go.etcd.io/bbolt v1.3.7/go.mod h1:N9Mkw9X8x5fupy0IKsmuqVtoGDyxsaDlbk4Rd05IAQw=
 golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI=
 golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
 golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=

backend/main.go

@@ -54,6 +54,12 @@ func main() {
 	}
 
 	controllers.GlobalJobQueue = controllers.NewJobQueue()
+	if controllers.GlobalJobQueue != nil && controllers.GlobalJobQueue.GetPersistentQueue() != nil {
+ maintenanceWorker := utils.NewMaintenanceWorker(controllers.GlobalJobQueue.GetPersistentQueue())
+ if err := maintenanceWorker.Start(); err != nil {
+ utils.Logger.Errorf("Failed to start maintenance worker: %v", err)
+ }
+	}
 	// OAuth2 client credentials
 	clientID := os.Getenv("CLIENT_ID")
 	clientSecret := os.Getenv("CLIENT_SEC")

backend/utils/maintenance_worker.go

@@ -0,0 +1,66 @@
+package utils
+
+import (
+	"os"
+	"strconv"
+
+	"github.com/robfig/cron/v3"
+)
+
+type MaintenanceWorker struct {
+	cron *cron.Cron
+	queue PersistentJobQueue
+}
+
+func NewMaintenanceWorker(queue PersistentJobQueue) *MaintenanceWorker {
+	return &MaintenanceWorker{
+ cron: cron.New(),
+ queue: queue,
+	}
+}
+
+func (mw *MaintenanceWorker) Start() error {
+	// CLEANUP_CRON_SCHEDULE: Cron expression for cleanup schedule
+	// Format: "minute hour day month weekday"
+	// Examples:
+	// "0 0 * * *" - Daily at midnight (default)
+	// "0 */6 * * *" - Every 6 hours
+	// "0 2 * * *" - Daily at 2 AM
+	// "0 0 * * 0" - Weekly on Sunday
+	schedule := os.Getenv("CLEANUP_CRON_SCHEDULE")
+	if schedule == "" {
+ schedule = "0 0 * * *"
+	}
+
+	// CLEANUP_RETENTION_DAYS: Number of days to keep completed/failed job logs
+	// Default: 7 days
+	// Set to higher value to keep logs longer, lower to cleanup more frequently
+	retentionDaysStr := os.Getenv("CLEANUP_RETENTION_DAYS")
+	retentionDays := 7
+	if retentionDaysStr != "" {
+ if days, err := strconv.Atoi(retentionDaysStr); err == nil {
+ retentionDays = days
+ }
+	}
+
+	_, err := mw.cron.AddFunc(schedule, func() {
+ Logger.Infof("Starting job cleanup, retention: %d days", retentionDays)
+ if err := mw.queue.CleanupOldJobs(retentionDays); err != nil {
+ Logger.Errorf("Failed to cleanup old jobs: %v", err)
+ } else {
+ Logger.Infof("Job cleanup completed successfully")
+ }
+	})
+
+	if err != nil {
+ return err
+	}
+
+	mw.cron.Start()
+	Logger.Infof("Maintenance worker started with schedule: %s", schedule)
+	return nil
+}
+
+func (mw *MaintenanceWorker) Stop() {
+	mw.cron.Stop()
+}