import { createClient } from 'redis';
import { Queue } from '@nivalis/n-queue';

// Define your payload schema
type MyPayload = {
  emailQueue: {
    sendEmail: {
      to: string;
      subject: string;
      body: string;
    };
    sendNotification: {
      userId: string;
      message: string;
    };
  };
};

// Create a Redis client factory
const getRedisClient = async () => {
  const client = createClient({
    url: process.env.REDIS_URL ?? 'redis://localhost:6379',
  });

  if (!client.isOpen) {
    await client.connect();
  }

  return client;
};

// Create a queue with concurrency limit
const emailQueue = new Queue<MyPayload, 'emailQueue'>(
  'emailQueue',
  getRedisClient,
  { concurrency: 5 }
);

// Add jobs to the queue
const emailJob = await emailQueue.add('sendEmail', {
  to: 'user@example.com',
  subject: 'Welcome!',
  body: 'Welcome to our platform!'
});

// Process a single job (automatic completion/failure handling)
await emailQueue.process(async (job) => {
  console.log(`Processing ${job.name} job ${job.id}`);
  await sendEmail(job.payload);
}, {});

// Or process a specific job type
await emailQueue.process(async (job) => {
  await sendNotification(job.payload);
}, { jobName: 'sendNotification' });

// Stream jobs in real-time (runs until you stop the process)
await emailQueue.stream(async (job) => {
  console.log(`Processing streamed job ${job.id}`);
  await processJob(job);
});

// Listen for events
emailQueue.on('saved', (jobId) => {
  console.log(`Job ${jobId} was saved`);
});

emailQueue.on('completed', (jobId) => {
  console.log(`Job ${jobId} completed successfully`);
});

import { Effect } from 'effect';
import { Job, Queue } from '@nivalis/n-queue';

const enqueueEmail = (queue: Queue<MyPayload, 'emailQueue'>) =>
  Effect.gen(function* () {
    const job = new Job({
      queue,
      name: 'sendEmail',
      payload: {
        to: 'user@example.com',
        subject: 'Welcome',
        body: '👋',
      },
    });

    return yield* job.saveEffect();
  });

class Queue<Payload, QueueName> {
  constructor(
    name: QueueName,
    getRedisClient: () => Promise<RedisClientType>,
    options?: QueueOptions
  );

  // Job Management
  add<JobName>(jobName: JobName, payload: Payload[QueueName][JobName]): Promise<Job>;
  process(
    fn: (job: Job) => Promise<void>,
    options: { jobName?: JobName; jobId?: string }
  ): Promise<void>;
  stream(fn: (job: Job) => Promise<void>, jobName?: JobName): Promise<void>;

  // Event Handling
  on(event: string, handler: (jobId: string) => void): void;
  once(event: string, handler: (jobId: string) => void): void;

  // Queue Information
  getStats(): Promise<{
    waiting: number;
    active: number;
    failed: number;
    completed: number;
  }>;
}

class Job<Payload, QueueName, JobName> {
  readonly id: string;
  readonly name: JobName;
  readonly state: JobState;
  readonly payload: Payload[QueueName][JobName];
  readonly createdAt: string;
  readonly updatedAt: string;

  progress: number;
  processedAt: string | null;
  attempts: number;
  failedReason: string | null;
  stacktrace: string[];

  // Job Operations
  save(): Promise<Job>;
  move(state: JobState): Promise<Job>;
  withState(state: JobState): Job;
}

type QueueOptions = {
  // Max concurrent jobs (-1 for unlimited)
  concurrency?: number;

  // Job visibility / stall recovery
  visibilityTimeoutMs?: number;
  stallCheckIntervalMs?: number;

  // Retry for failed jobs
  retry?: {
    maxAttempts?: number;
    backoffStrategy?: {
      initialDelay: number;
      factor: number;
      maxDelay: number;
      jitter?: number;
    };
  };

  // Optional integrations
  logger?: {
    debug?: (msg: string, meta?: unknown) => void;
    info?: (msg: string, meta?: unknown) => void;
    warn?: (msg: string, meta?: unknown) => void;
    error?: (msg: string, meta?: unknown) => void;
  };
  metrics?: {
    increment?: (name: string, labels?: Record<string, string>) => void;
    observe?: (name: string, value: number, labels?: Record<string, string>) => void;
  };
};

type JobState = 'waiting' | 'active' | 'completed' | 'failed';