package main

import (
	"context"
	"log"
	"os/signal"
	"syscall"
	"time"

	"github.com/gnikyt/cq/v2"
)

func doWork(ctx context.Context) error {
	meta := cq.MetaFromContext(ctx)
	log.Printf("job %s started, queued %v ago", meta.ID, time.Since(meta.EnqueuedAt))
	time.Sleep(2 * time.Second)
	log.Printf("job %s completed", meta.ID)
	return nil
}

func main() {
	// Listen for interrupt signals.
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	// Create queue with the signal context.
	queue := cq.NewQueue(1, 10, 100, cq.WithContext(ctx))
	queue.Start()

	// Submit work...
	_, _ = queue.Submit(context.Background(), func(ctx context.Context) error {
		return doWork(ctx)
	})

	// Wait for shutdown signal.
	<-ctx.Done()

	// Stop queue, wait for in-flight jobs to finish.
	queue.Stop(true)
}

highQ := cq.NewQueue(5, 50, 1000) // High-priority lane.
lowQ := cq.NewQueue(1, 5, 5000)   // Bulk/background lane.

mgr := cq.NewQueueManager()
if err := mgr.Register("high", highQ); err != nil {
	log.Fatal(err)
}
if err := mgr.Register("low", lowQ); err != nil {
	log.Fatal(err)
}

mgr.StartAll()
defer mgr.StopAll(true)

if _, err := mgr.Submit(ctx, "high", processCritical); err != nil {
	log.Fatal(err)
}
if _, err := mgr.Submit(ctx, "low", processBulk); err != nil {
	log.Fatal(err)
}

if _, err := mgr.SubmitAfter(ctx, "low", processLater, 30*time.Second); err != nil {
	log.Fatal(err)
}

criticalBase := cq.NewQueue(5, 20, 500)
criticalBase.Start()

bulkBase := cq.NewQueue(2, 10, 1000)
bulkBase.Start()

pmgr := cq.NewPriorityQueueManager()
if err := pmgr.Register("critical", cq.NewPriorityQueue(criticalBase, 100)); err != nil {
	log.Fatal(err)
}
if err := pmgr.Register("bulk", cq.NewPriorityQueue(bulkBase, 200)); err != nil {
	log.Fatal(err)
}

defer pmgr.StopAll(true)

if _, err := pmgr.Submit(ctx, "critical", processNow, cq.PriorityHighest); err != nil {
	log.Fatal(err)
}
if _, err := pmgr.SubmitAfter(ctx, "bulk", processLater, cq.PriorityLow, time.Minute); err != nil {
	log.Fatal(err)
}

job := WithOutcome(              // 3. Outermost: catches final outcome.
	WithRetryPolicy(            // 2. Preferred retry wrapper.
		WithTimeout(             // 1. Innermost: runs with timeout.
				actualJob,
				5*time.Minute,
			),
		RetryPolicy{
			MaxAttempts: 3,
			Backoff:     ExponentialBackoff,
		},
	),
	onComplete,
	onFail,
	onDiscard,
)

job := cq.WithRetryPolicy(
	cq.WithTimeout(fetchFromAPI, 10*time.Second),
	cq.RetryPolicy{
		MaxAttempts: 3,
		Backoff:     cq.ExponentialBackoff,
	},
)
_, _ = queue.Submit(context.Background(), job)

store := cq.NewMemoryCheckpointStore()
step := cq.WithCheckpoint(
	sendInvoice,
	"send-invoice",
	store,
	cq.WithCheckpointNamespace("billing"),
)

job := cq.WithChain(
	validateOrder,
	step, // Will be skipped on retry after first success.
	notifyCustomer,
)
_, _ = queue.Submit(context.Background(), job)

if err := cq.SaveCheckpointData(ctx, []byte("batch-42-complete")); err != nil {
	return err
}

locker := cq.NewUniqueMemoryLocker()
job := cq.WithUnique(
	cq.WithTimeout(processOrder, 30*time.Second),
	"order:123",
	5*time.Minute,
	locker,
)
_, _ = queue.Submit(context.Background(), job)

job := cq.WithUnique(func(ctx context.Context) error {
	if err := cq.TouchLock(ctx, 30*time.Second); err != nil {
		return err // Handle cq.ErrUniqueLeaseLost / cq.ErrTouchLockUnavailable as needed.
	}
	return doWork(ctx)
}, "index-products", 30*time.Second, locker)

scheduler := cq.NewScheduler(context.Background(), queue)
defer scheduler.Stop()

schedule, err := scheduler.Every(
	"sync-products",
	10*time.Minute,
	syncProductsJob,
	cq.WithJobName("sync-products"),
)

latest, submitErr, attempted := schedule.Latest()

queue := cq.NewQueue(1, 100, 1000)
queue.Start()
defer queue.Stop(true)

// Recommended v2 submission API.
handle, err := queue.Submit(ctx, job,
	cq.WithJobID("message-123"),
	cq.WithJobName("process-message"),
	cq.WithJobAttribute("source", "sqs"),
)
if err != nil {
	log.Fatal(err) // Job was not accepted.
}

// Waiting is optional. A wait timeout does not cancel the running job.
if err := handle.Wait(ctx); err != nil {
	log.Printf("job failed or wait ended: %v", err)
}

// Blocks until accepted or ctx ends.
handle, err := queue.Submit(ctx, job)

// Returns ErrQueueFull instead of waiting for capacity.
handle, err = queue.Submit(ctx, job, cq.WithNonBlocking())

scheduled, err := queue.SubmitAfter(ctx, job, 2*time.Minute)
handles, err := queue.SubmitBatch(ctx, jobs)
scheduledHandles, err := queue.SubmitBatchAfter(ctx, jobs, 30*time.Second)

// Resubmit a running job later.
rescheduled, err := cq.Reschedule(ctx, queue, job, time.Minute, cq.RescheduleReasonManualRetry)

queue.RunningWorkers()           // Current running workers.
queue.IdleWorkers()              // Current idle workers.
queue.Capacity()                 // Job channel capacity.
queue.WorkerRange()              // (min, max) workers.
queue.Stats()                    // Single-call queue snapshot.
queue.SetWorkerRange(2, 20)      // Update (min, max) at runtime.
queue.TallyOf(cq.JobStateFailed) // Count by state.

// Available job states for TallyOf:
// cq.JobStateCreated   - Total jobs accepted.
// cq.JobStatePending   - Jobs waiting in the queue.
// cq.JobStateActive    - Jobs currently executing.
// cq.JobStateFailed    - Jobs completed with error.
// cq.JobStateCancelled - Jobs completed through handle cancellation.
// cq.JobStateCompleted - Jobs completed successfully.
// cq.JobStateDiscarded - Jobs marked as discarded outcomes.

if err := queue.SetWorkerRange(2, 20); err != nil {
	log.Fatal(err)
}

queue := cq.NewQueue(1, 10, 100,
	cq.WithWorkerIdleTick(500*time.Millisecond),
	cq.WithContext(ctx),
	cq.WithPanicHandler(func(err any) {
		log.Printf("panic: %v", err)
	}),
)

// Available options:
// cq.WithWorkerIdleTick(d)           - Interval for idle worker cleanup (default 5s).
// cq.WithContext(ctx)                - Parent context for the queue.
// cq.WithCancelableContext(ctx, fn)  - Parent context with custom cancel function.
// cq.WithPanicHandler(fn)            - Custom handler override for job panics.
// cq.WithIDGenerator(fn)             - Override fallback job ID generation.
// cq.WithQueueName(name)             - Stable queue name for observability events/stats.
// cq.WithPauseStore(store, key)      - Share pause state across queue instances.
// cq.WithPausePollTick(d)            - Poll interval for distributed pause sync.
// cq.WithPauseBehavior(mode)         - Buffer or reject enqueue while paused.
// cq.WithMiddleware(mw...)           - Apply queue-level wrappers to all jobs.
// cq.WithHooks(hooks)                - Register queue lifecycle hooks.

withLogging := func(next cq.Job) cq.Job {
	return func(ctx context.Context) error {
		log.Println("job start")
		err := next(ctx)
		log.Printf("job end: %v", err)
		return err
	}
}

queue := cq.NewQueue(1, 10, 100, cq.WithMiddleware(withLogging))

if err := queue.Pause(); err != nil {
	log.Fatal(err)
}

// ... perform maintenance ...

if err := queue.Resume(); err != nil {
	log.Fatal(err)
}

queue.Stop(true)   // Wait for jobs to finish.
queue.Stop(false)  // Don't wait for queued jobs.
queue.StopContext(ctx) // Wait for jobs or stop when ctx is done.
queue.StopTimeout(5 * time.Second) // Wait up to timeout duration.
queue.Terminate()  // Immediate shutdown.