/**
 * @beignet/provider-locks-redis
 *
 * Redis-backed locks provider for Beignet.
 */

import type {
  LeaseAcquireOptions,
  LeaseHandle,
  LeaseRenewOptions,
  LocksPort,
} from "@beignet/core/locks";
import { LeaseOptionsError } from "@beignet/core/locks";
import {
  createProvider,
  createProviderInstrumentation,
  type ProviderInstrumentationTarget,
} from "@beignet/core/providers";
import { Redis } from "ioredis";
import { z } from "zod";

const DEFAULT_CONNECT_TIMEOUT_MS = 5000;
const DEFAULT_MAX_RETRIES_PER_REQUEST = 2;
const DEFAULT_CONNECT_MAX_ATTEMPTS = 3;
const DEFAULT_PREFIX = "beignet:locks";
const DEFAULT_RETRY_DELAY_MS = 50;
const RECONNECT_BACKOFF_CAP_MS = 2000;

const ACQUIRE_SCRIPT = `
if redis.call("set", KEYS[1], ARGV[1], "PX", ARGV[2], "NX") then
  return redis.call("incr", KEYS[2])
end
return nil
`;

const RELEASE_SCRIPT = `
if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("del", KEYS[1])
end
return 0
`;

const RENEW_SCRIPT = `
if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("pexpire", KEYS[1], ARGV[2])
end
return 0
`;

const NumberString = z
  .string()
  .regex(/^\d+$/, "Expected a non-negative integer string")
  .transform((value) => Number.parseInt(value, 10));

const RedisLocksConfigSchema = z.object({
  URL: z.string().url(),
  DB: NumberString.optional(),
  PREFIX: z.string().default(DEFAULT_PREFIX),
  CONNECT_TIMEOUT_MS: NumberString.default(DEFAULT_CONNECT_TIMEOUT_MS),
  MAX_RETRIES_PER_REQUEST: NumberString.default(
    DEFAULT_MAX_RETRIES_PER_REQUEST,
  ),
  CONNECT_MAX_ATTEMPTS: NumberString.default(DEFAULT_CONNECT_MAX_ATTEMPTS),
});

export type RedisLocksConfig = z.infer<typeof RedisLocksConfigSchema>;

/**
 * Redis client surface used by the locks adapter.
 */
export interface RedisLocksClient {
  eval(
    script: string,
    numberOfKeys: number,
    ...args: unknown[]
  ): Promise<unknown>;
  del(key: string): Promise<number>;
  ping?(): Promise<string>;
}

export interface RedisLocksHealth {
  ok: boolean;
  message?: string;
  metadata: {
    adapter: "redis-locks";
    prefix: string;
    response?: string;
  };
}

/**
 * Escape hatch for apps that need the raw Redis client.
 */
export interface RedisLocksEscapeHatch {
  client: RedisLocksClient;
  prefix: string;
  checkHealth(): Promise<RedisLocksHealth>;
}

/**
 * Ports contributed by the Redis locks provider.
 */
export interface RedisLocksProviderPorts {
  locks: LocksPort;
  redisLocks: RedisLocksEscapeHatch;
}

/**
 * Options for the direct Redis locks factory.
 */
export interface CreateRedisLocksOptions {
  client: RedisLocksClient;
  prefix?: string;
  instrumentation?: ProviderInstrumentationTarget;
  createOwnerToken?: () => string;
  sleep?: (ms: number) => Promise<void>;
}

/**
 * Options for the Redis locks lifecycle provider.
 */
export interface CreateRedisLocksProviderOptions {
  client?: RedisLocksClient;
  prefix?: string;
  connectTimeoutMs?: number;
  maxRetriesPerRequest?: number;
  retryStrategy?: (times: number) => number | null;
  db?: number;
}

/**
 * Create a Redis-backed locks port.
 */
export function createRedisLocks(options: CreateRedisLocksOptions): LocksPort {
  const client = options.client;
  const prefix = options.prefix ?? DEFAULT_PREFIX;
  const sleep = options.sleep ?? defaultSleep;
  const createOwnerToken = options.createOwnerToken ?? createRandomToken;
  const instrumentation = createProviderInstrumentation(
    options.instrumentation,
    {
      providerName: "locks-redis",
      watcher: "locks",
    },
  );

  const fullKey = (key: string) => (prefix ? `${prefix}:${key}` : key);
  const fencingKey = (key: string) => `${fullKey(key)}:fence`;

  const locks: LocksPort = {
    async acquire(key, acquireOptions) {
      validateAcquireOptions(key, acquireOptions);
      const startedAt = Date.now();
      const waitMs = acquireOptions.waitMs ?? 0;
      const retryDelayMs =
        acquireOptions.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS;
      const deadline = Date.now() + waitMs;
      const ownerToken = acquireOptions.ownerToken ?? createOwnerToken();

      while (true) {
        const fencingToken = await client.eval(
          ACQUIRE_SCRIPT,
          2,
          fullKey(key),
          fencingKey(key),
          ownerToken,
          acquireOptions.ttlMs,
        );

        if (fencingToken !== null && fencingToken !== undefined) {
          const lease = createRedisLease({
            client,
            key,
            redisKey: fullKey(key),
            ownerToken,
            ttlMs: acquireOptions.ttlMs,
            fencingToken: Number(fencingToken),
          });
          recordAcquire({
            instrumentation,
            key,
            acquired: true,
            ttlMs: acquireOptions.ttlMs,
            waitMs,
            durationMs: Date.now() - startedAt,
            metadata: acquireOptions.metadata,
          });
          return { acquired: true, lease };
        }

        if (waitMs <= 0) {
          recordAcquire({
            instrumentation,
            key,
            acquired: false,
            reason: "unavailable",
            ttlMs: acquireOptions.ttlMs,
            waitMs,
            durationMs: Date.now() - startedAt,
            metadata: acquireOptions.metadata,
          });
          return { acquired: false, reason: "unavailable" };
        }

        const remainingMs = deadline - Date.now();
        if (remainingMs <= 0) {
          recordAcquire({
            instrumentation,
            key,
            acquired: false,
            reason: "timeout",
            ttlMs: acquireOptions.ttlMs,
            waitMs,
            durationMs: Date.now() - startedAt,
            metadata: acquireOptions.metadata,
          });
          return { acquired: false, reason: "timeout" };
        }

        await sleep(Math.min(retryDelayMs, remainingMs));
      }
    },
    async withLease(key, acquireOptions, fn) {
      const result = await locks.acquire(key, acquireOptions);
      if (!result.acquired) return undefined;

      try {
        return await fn({ lease: result.lease });
      } finally {
        await result.lease.release();
      }
    },
    restore(key, ownerToken) {
      if (!key) throw new LeaseOptionsError("Lease key is required.");
      if (!ownerToken) {
        throw new LeaseOptionsError("Lease owner token is required.");
      }

      return createRedisLease({
        client,
        key,
        redisKey: fullKey(key),
        ownerToken,
        ttlMs: 1,
      });
    },
    async forceRelease(key) {
      if (!key) throw new LeaseOptionsError("Lease key is required.");
      const released = (await client.del(fullKey(key))) > 0;
      instrumentation.custom({
        name: "locks.forceRelease",
        label: "Lock force release",
        summary: released ? "Lease force released" : "Lease not found",
        details: { key, prefix, released },
      });
      return released;
    },
  };

  return locks;
}

async function checkRedisLocksHealth(
  client: RedisLocksClient,
  prefix: string,
): Promise<RedisLocksHealth> {
  try {
    const response =
      typeof client.ping === "function"
        ? await client.ping()
        : String(await client.eval('return "PONG"', 0));

    return {
      ok: true,
      metadata: { adapter: "redis-locks", prefix, response },
    };
  } catch (error) {
    return {
      ok: false,
      message: errorMessage(error),
      metadata: { adapter: "redis-locks", prefix },
    };
  }
}

/**
 * Create an env-backed Redis locks provider.
 */
export function createRedisLocksProvider(
  options: CreateRedisLocksProviderOptions = {},
) {
  const ConfigSchema = RedisLocksConfigSchema.extend({
    DB:
      options.db !== undefined
        ? NumberString.default(options.db)
        : RedisLocksConfigSchema.shape.DB,
    PREFIX:
      options.prefix !== undefined
        ? z.string().default(options.prefix)
        : RedisLocksConfigSchema.shape.PREFIX,
    CONNECT_TIMEOUT_MS:
      options.connectTimeoutMs !== undefined
        ? NumberString.default(options.connectTimeoutMs)
        : RedisLocksConfigSchema.shape.CONNECT_TIMEOUT_MS,
    MAX_RETRIES_PER_REQUEST:
      options.maxRetriesPerRequest !== undefined
        ? NumberString.default(options.maxRetriesPerRequest)
        : RedisLocksConfigSchema.shape.MAX_RETRIES_PER_REQUEST,
  });

  const providerMetadata = {
    packageName: "@beignet/provider-locks-redis",
    ports: ["locks", "redisLocks"],
    watchers: ["locks"],
  } as const;

  if (options.client) {
    const client = options.client;

    return createProvider({
      name: "locks-redis",
      metadata: providerMetadata,
      async setup({ ports }) {
        const configuredPrefix = options.prefix ?? DEFAULT_PREFIX;

        return {
          ports: {
            locks: createRedisLocks({
              client,
              prefix: configuredPrefix,
              instrumentation: ports,
            }),
            redisLocks: {
              client,
              prefix: configuredPrefix,
              checkHealth: () =>
                checkRedisLocksHealth(client, configuredPrefix),
            },
          } satisfies RedisLocksProviderPorts,
        };
      },
    });
  }

  return createProvider({
    name: "locks-redis",
    metadata: providerMetadata,
    config: {
      schema: ConfigSchema,
      envPrefix: "REDIS_LOCKS_",
    },
    async setup({ ports, config }) {
      const configuredPrefix =
        config?.PREFIX ?? options.prefix ?? DEFAULT_PREFIX;

      if (!config) {
        throw new Error(
          "[redisLocksProvider] Missing Redis locks config. " +
            "Please set REDIS_LOCKS_URL environment variable.",
        );
      }

      const client = await connectRedisClient(config, options);

      return {
        ports: {
          locks: createRedisLocks({
            client,
            prefix: configuredPrefix,
            instrumentation: ports,
          }),
          redisLocks: {
            client,
            prefix: configuredPrefix,
            checkHealth: () => checkRedisLocksHealth(client, configuredPrefix),
          },
        } satisfies RedisLocksProviderPorts,
        async stop() {
          try {
            await client.quit();
          } catch {
            // Ignore shutdown errors.
          }
        },
      };
    },
  });
}

/**
 * Default env-backed Redis locks provider.
 */
export const redisLocksProvider = createRedisLocksProvider();

function createRedisLease(args: {
  client: RedisLocksClient;
  key: string;
  redisKey: string;
  ownerToken: string;
  ttlMs: number;
  fencingToken?: number;
}): LeaseHandle {
  const lease: LeaseHandle = {
    key: args.key,
    ownerToken: args.ownerToken,
    expiresAt: new Date(Date.now() + args.ttlMs),
    fencingToken: args.fencingToken,
    async renew(options?: LeaseRenewOptions) {
      const ttlMs = options?.ttlMs ?? args.ttlMs;
      if (!Number.isFinite(ttlMs) || ttlMs <= 0) {
        throw new LeaseOptionsError("Lease ttlMs must be a positive number.");
      }

      const renewed = Number(
        await args.client.eval(
          RENEW_SCRIPT,
          1,
          args.redisKey,
          args.ownerToken,
          ttlMs,
        ),
      );
      if (renewed !== 1) return false;
      args.ttlMs = ttlMs;
      lease.expiresAt = new Date(Date.now() + ttlMs);
      return true;
    },
    async release() {
      const released = Number(
        await args.client.eval(
          RELEASE_SCRIPT,
          1,
          args.redisKey,
          args.ownerToken,
        ),
      );
      return released === 1;
    },
  };

  return lease;
}

async function connectRedisClient(
  config: RedisLocksConfig,
  options: CreateRedisLocksProviderOptions,
): Promise<Redis> {
  const connectMaxAttempts =
    config.CONNECT_MAX_ATTEMPTS ?? DEFAULT_CONNECT_MAX_ATTEMPTS;
  let connectedOnce = false;

  const defaultRetryStrategy = (attempt: number): number | null => {
    if (!connectedOnce) {
      if (attempt >= connectMaxAttempts) return null;
      return reconnectBackoff(attempt);
    }
    return reconnectBackoff(attempt);
  };

  const client = new Redis(config.URL, {
    db: config.DB ?? options.db ?? 0,
    lazyConnect: true,
    connectTimeout:
      config.CONNECT_TIMEOUT_MS ??
      options.connectTimeoutMs ??
      DEFAULT_CONNECT_TIMEOUT_MS,
    maxRetriesPerRequest:
      config.MAX_RETRIES_PER_REQUEST ??
      options.maxRetriesPerRequest ??
      DEFAULT_MAX_RETRIES_PER_REQUEST,
    retryStrategy: options.retryStrategy ?? defaultRetryStrategy,
  });

  try {
    await client.connect();
    connectedOnce = true;
  } catch (error) {
    client.disconnect();
    throw new Error(
      `[redisLocksProvider] Failed to connect to Redis at ${redactConnectionUrl(
        config.URL,
      )}: ${errorMessage(error)}`,
    );
  }

  return client;
}

function recordAcquire(args: {
  instrumentation: ReturnType<typeof createProviderInstrumentation>;
  key: string;
  acquired: boolean;
  reason?: "unavailable" | "timeout";
  ttlMs: number;
  waitMs: number;
  durationMs: number;
  metadata?: LeaseAcquireOptions["metadata"];
}) {
  args.instrumentation.custom({
    name: "locks.acquire",
    label: "Lock acquire",
    summary: args.acquired ? "Lease acquired" : "Lease not acquired",
    details: {
      key: args.key,
      acquired: args.acquired,
      reason: args.reason,
      ttlMs: args.ttlMs,
      waitMs: args.waitMs,
      durationMs: args.durationMs,
      metadata: args.metadata,
    },
  });
}

function validateAcquireOptions(key: string, options: LeaseAcquireOptions) {
  if (!key) throw new LeaseOptionsError("Lease key is required.");
  if (!Number.isFinite(options.ttlMs) || options.ttlMs <= 0) {
    throw new LeaseOptionsError("Lease ttlMs must be a positive number.");
  }
  if (options.waitMs !== undefined && options.waitMs < 0) {
    throw new LeaseOptionsError("Lease waitMs must be zero or greater.");
  }
  if (options.retryDelayMs !== undefined && options.retryDelayMs <= 0) {
    throw new LeaseOptionsError("Lease retryDelayMs must be positive.");
  }
}

function reconnectBackoff(attempt: number): number {
  return Math.min(2 ** attempt * 100, RECONNECT_BACKOFF_CAP_MS);
}

function redactConnectionUrl(url: string): string {
  try {
    const parsed = new URL(url);
    if (parsed.username) parsed.username = "redacted";
    if (parsed.password) parsed.password = "redacted";
    if (parsed.search) parsed.search = "?redacted";
    parsed.hash = "";
    return parsed.toString();
  } catch {
    return "[invalid URL]";
  }
}

function createRandomToken(): string {
  if (typeof crypto !== "undefined" && "randomUUID" in crypto) {
    return crypto.randomUUID();
  }

  return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`;
}

function defaultSleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function errorMessage(error: unknown): string {
  return error instanceof Error ? error.message : String(error);
}