import { Redis, SentinelConnectionOptions } from 'ioredis';
import { registerShutdownHandler } from '../utils/platformEventHandling.js';
import Sentry from './sentry.js';

let redisClient: Redis | null = null;

class AttemptsExhaustedError extends Error {}

const retryStrategy = (attempts) => {
  if (attempts >= 10) {
    throw new AttemptsExhaustedError('Redis server reconnect attempts exhausted');
  }
  return attempts * 500;
};

const constructRedisClient = (commandTimeout?: number): Redis => {
  const sentinelOpts: SentinelConnectionOptions = {
    name: process.env.REDIS_SENTINEL_NAME,
    sentinels: [
      {
        host: process.env.REDIS_SENTINEL_HOST,
        port: Number.parseInt(process.env.REDIS_SENTINEL_PORT),
      },
    ],
    sentinelUsername: process.env.REDIS_USERNAME,
    sentinelPassword: process.env.REDIS_PASSWORD,
    role: 'master',
    sentinelRetryStrategy: retryStrategy,
    sentinelCommandTimeout: 2000,
    updateSentinels: true,
    failoverDetector: true,
  };
  const standaloneOpts = {
    host: process.env.REDIS_HOST,
    port: Number.parseInt(process.env.REDIS_PORT),
  };
  const client = new Redis({
    retryStrategy,
    username: process.env.REDIS_USERNAME,
    password: process.env.REDIS_PASSWORD,
    connectTimeout: 2000,
    disconnectTimeout: 2000,
    noDelay: true,
    commandTimeout,
    ...(process.env.REDIS_SENTINEL_NAME ? sentinelOpts : standaloneOpts),
  });
  client.on('error', (error) => {
    if (error instanceof AttemptsExhaustedError) {
      Sentry.captureException(error);
      throw error;
    }
    console.error(error);
  });
  return client;
};

// lazy init redis client
export const getRedisClient = (): Redis => {
  if (redisClient) {
    return redisClient;
  }
  redisClient = constructRedisClient();
  registerShutdownHandler(() => {
    redisClient?.disconnect();
  });

  return redisClient;
};

export const maybeGetRedisClient = (): Redis | null => {
  return redisClient;
};

export const withIsolatedRedisClient = async <T>(
  commandTimeout: number,
  fn: (clientPromise: Promise<Redis>) => Promise<T>
): Promise<T> => {
  const clientPromise = (async () => constructRedisClient(commandTimeout))();
  try {
    return await fn(clientPromise);
  } finally {
    (await clientPromise).disconnect();
  }
};

const randomId = () => Math.random().toString(36).substring(2, 9);

// enqueue data and wait for a response
export const enqueueWithTimeout = async (queue: string, data: any, timeout: number) => {
  const mainClient = getRedisClient();
  return await withIsolatedRedisClient(timeout * 1000 + 2000, async (clientPromise: Promise<Redis>) => {
    const id = randomId();
    const now = Date.now() / 1000.0;
    await mainClient.lpush(queue, JSON.stringify({ id, data, timeSubmitted: now, timesOutAt: now + timeout }));
    const blockingClient = await clientPromise;
    const res = await blockingClient.blpop(id, timeout);
    if (!res) {
      throw new Error('Timeout waiting for response');
    }
    return JSON.parse(res[1]);
  });
};

// enqueue data without waiting for a response
export const enqueue = async (queue: string, data: any, environment: string = process.env.ENVIRONMENT) => {
  const now = Date.now() / 1000.0;
  const key = environment ? `${queue}-${environment}` : queue;
  await getRedisClient().lpush(key, JSON.stringify({ data, timeSubmitted: now }));
};

const LIVENESS_POLL_INTERVAL = 10000; // ms
const LIVENESS_EXPIRE = Math.round((2 * LIVENESS_POLL_INTERVAL) / 1000);
const BLOCK_TIMEOUT = 5; // seconds

export const handleQueue = async (
  queue: string,
  handler: (data: any) => Promise<any>,
  environment: string = process.env.ENVIRONMENT
) => {
  const workerId = randomId();
  const key = environment ? `${queue}-${environment}` : queue;
  const livenessKey = `worker:${key}:${workerId}`;
  const redriveKey = `work:${key}:${workerId}`;
  const mainClient = getRedisClient();
  let quit = false;
  const livenessThunk = async () => {
    try {
      await mainClient.set(livenessKey, 'alive', 'EX', LIVENESS_EXPIRE);
    } catch (e) {
      console.error(e);
    }
    if (quit) {
      return;
    }
    setTimeout(livenessThunk, LIVENESS_POLL_INTERVAL);
  };
  await mainClient.set(livenessKey, 'alive', 'EX', LIVENESS_EXPIRE);
  setTimeout(livenessThunk, LIVENESS_POLL_INTERVAL);
  setImmediate(async () => {
    try {
      await withIsolatedRedisClient(BLOCK_TIMEOUT * 1000 + 2000, async (clientPromise: Promise<Redis>) => {
        const blockingClient = await clientPromise;
        while (true) {
          const work = await blockingClient.brpoplpush(key, redriveKey, BLOCK_TIMEOUT);
          if (work === null) {
            continue;
          }
          try {
            const { id, data } = JSON.parse(work);
            const result = await handler(data);
            if (id !== undefined) {
              await blockingClient.multi().del(redriveKey).lpush(id, JSON.stringify(result)).exec();
            } else {
              await blockingClient.del(redriveKey);
            }
          } catch (e) {
            console.error(e);
            Sentry.captureException(e);
            await blockingClient.del(redriveKey);
          }
        }
      });
    } finally {
      quit = true;
    }
  });
};