/**
 * @beignet/provider-event-bus-redis
 *
 * Redis Pub/Sub event bus provider that implements EventBusPort for
 * cross-process, best-effort domain event delivery.
 */

import type {
  DomainEventDef,
  EventBusPort,
  InferEventPayload,
} from "@beignet/core/ports";
import {
  type AnyProviderConfigSchema,
  createProvider,
  createProviderInstrumentation,
  type ProviderInstrumentationTarget,
  type ServiceProvider,
} 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_CHANNEL_PREFIX = "beignet:event-bus:";
const RECONNECT_BACKOFF_CAP_MS = 2000;

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

const RedisEventBusConfigSchema = z.object({
  URL: z.string().url(),
  DB: NumberString.optional(),
  CHANNEL_PREFIX: z.string().default(DEFAULT_CHANNEL_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),
}) satisfies z.ZodType<RedisEventBusConfig>;

/**
 * Validated configuration for the Redis event bus provider.
 */
export interface RedisEventBusConfig {
  URL: string;
  DB?: number;
  CHANNEL_PREFIX: string;
  CONNECT_TIMEOUT_MS: number;
  MAX_RETRIES_PER_REQUEST: number;
  CONNECT_MAX_ATTEMPTS: number;
}

export interface RedisEventBusPublisherClient {
  publish(channel: string, message: string): Promise<number> | number;
  ping?(): Promise<string> | string;
}

export interface RedisEventBusSubscriberClient {
  subscribe(...channels: string[]): Promise<unknown> | unknown;
  unsubscribe(...channels: string[]): Promise<unknown> | unknown;
  on(
    event: "message",
    handler: (channel: string, message: string) => void,
  ): unknown;
  off?(
    event: "message",
    handler: (channel: string, message: string) => void,
  ): unknown;
  removeListener?(
    event: "message",
    handler: (channel: string, message: string) => void,
  ): unknown;
  ping?(): Promise<string> | string;
}

export interface RedisEventBusHealth {
  ok: boolean;
  message?: string;
  metadata: {
    adapter: "redis-event-bus";
    channelPrefix: string;
    publisherResponse?: string;
    subscriberResponse?: string;
  };
}

export interface RedisEventBusEscapeHatch {
  publisher: RedisEventBusPublisherClient;
  subscriber: RedisEventBusSubscriberClient;
  channelPrefix: string;
  checkHealth(): Promise<RedisEventBusHealth>;
}

export interface RedisEventBusProviderPorts {
  eventBus: EventBusPort;
  redisEventBus: RedisEventBusEscapeHatch;
}

/**
 * Redis event bus lifecycle provider with stable contributed-port typing.
 */
export type RedisEventBusProvider = ServiceProvider<
  unknown,
  AnyProviderConfigSchema<RedisEventBusConfig>,
  Pick<RedisEventBusProviderPorts, keyof RedisEventBusProviderPorts>
>;

export interface CreateRedisEventBusOptions {
  publisher: RedisEventBusPublisherClient;
  subscriber: RedisEventBusSubscriberClient;
  channelPrefix?: string;
  instrumentation?: ProviderInstrumentationTarget;
  onHandlerError?: (
    error: unknown,
    eventName: string,
    payload: unknown,
  ) => void;
  onSubscriberError?: (error: unknown) => void;
}

export interface CreateRedisEventBusProviderOptions {
  channelPrefix?: string;
  connectTimeoutMs?: number;
  maxRetriesPerRequest?: number;
  retryStrategy?: (times: number) => number | null;
  db?: number;
}

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

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 reconnectBackoff(attempt: number): number {
  return Math.min(2 ** attempt * 100, RECONNECT_BACKOFF_CAP_MS);
}

function createRetryStrategy(connectMaxAttempts: number): {
  strategy: (attempt: number) => number | null;
  markConnected: () => void;
} {
  let connectedOnce = false;

  return {
    strategy(attempt) {
      if (!connectedOnce && attempt >= connectMaxAttempts) {
        return null;
      }
      return reconnectBackoff(attempt);
    },
    markConnected() {
      connectedOnce = true;
    },
  };
}

function channelForEvent(channelPrefix: string, eventName: string): string {
  return `${channelPrefix}${eventName}`;
}

function eventNameForChannel(
  channelPrefix: string,
  channel: string,
): string | undefined {
  if (channelPrefix === "") return channel;
  if (!channel.startsWith(channelPrefix)) return undefined;
  return channel.slice(channelPrefix.length);
}

function safeJsonParse(input: string): unknown {
  return JSON.parse(input) as unknown;
}

async function checkRedisEventBusHealth(
  publisher: RedisEventBusPublisherClient,
  subscriber: RedisEventBusSubscriberClient,
  channelPrefix: string,
): Promise<RedisEventBusHealth> {
  try {
    const [publisherResponse, subscriberResponse] = await Promise.all([
      publisher.ping ? Promise.resolve(publisher.ping()) : Promise.resolve(""),
      subscriber.ping
        ? Promise.resolve(subscriber.ping())
        : Promise.resolve(""),
    ]);

    return {
      ok: true,
      metadata: {
        adapter: "redis-event-bus",
        channelPrefix,
        ...(publisherResponse ? { publisherResponse } : {}),
        ...(subscriberResponse ? { subscriberResponse } : {}),
      },
    };
  } catch (error) {
    return {
      ok: false,
      message: errorMessage(error),
      metadata: {
        adapter: "redis-event-bus",
        channelPrefix,
      },
    };
  }
}

export function createRedisEventBus(
  options: CreateRedisEventBusOptions,
): EventBusPort {
  const channelPrefix = options.channelPrefix ?? DEFAULT_CHANNEL_PREFIX;
  const handlers = new Map<
    string,
    Set<(payload: unknown) => void | Promise<void>>
  >();
  const subscribedChannels = new Set<string>();
  const instrumentation = createProviderInstrumentation(
    options.instrumentation,
    {
      providerName: "event-bus-redis",
      watcher: "eventBus",
    },
  );

  const reportSubscriberError = (error: unknown) => {
    try {
      options.onSubscriberError?.(error);
    } catch {
      // Error callbacks should not create unhandled rejections.
    }
  };

  const reportHandlerError = (
    error: unknown,
    eventName: string,
    payload: unknown,
  ) => {
    instrumentation.custom({
      name: "eventBus.handler.failed",
      label: "Event handler failed",
      summary: `Handler failed for "${eventName}"`,
      details: {
        eventName,
        error: errorMessage(error),
      },
    });

    try {
      options.onHandlerError?.(error, eventName, payload);
    } catch {
      // Error callbacks should not create unhandled rejections.
    }
  };

  const messageHandler = (channel: string, message: string) => {
    const eventName = eventNameForChannel(channelPrefix, channel);
    if (!eventName) return;

    const eventHandlers = handlers.get(eventName);
    if (!eventHandlers || eventHandlers.size === 0) return;

    let payload: unknown;
    try {
      const parsed = safeJsonParse(message);
      if (
        typeof parsed !== "object" ||
        parsed === null ||
        !("payload" in parsed)
      ) {
        throw new Error("Redis event bus message is missing a payload field.");
      }
      payload = (parsed as { payload: unknown }).payload;
    } catch (error) {
      reportSubscriberError(error);
      return;
    }

    instrumentation.custom({
      name: "eventBus.received",
      label: "Event received",
      summary: `Received "${eventName}"`,
      details: { eventName },
    });

    for (const handler of [...eventHandlers]) {
      try {
        Promise.resolve(handler(payload)).catch((error: unknown) => {
          reportHandlerError(error, eventName, payload);
        });
      } catch (error) {
        reportHandlerError(error, eventName, payload);
      }
    }
  };

  options.subscriber.on("message", messageHandler);

  const ensureSubscribed = (eventName: string) => {
    const channel = channelForEvent(channelPrefix, eventName);
    if (subscribedChannels.has(channel)) return;
    subscribedChannels.add(channel);
    Promise.resolve(options.subscriber.subscribe(channel)).catch((error) => {
      subscribedChannels.delete(channel);
      reportSubscriberError(error);
    });
  };

  const unsubscribeChannel = (eventName: string) => {
    const channel = channelForEvent(channelPrefix, eventName);
    if (!subscribedChannels.delete(channel)) return;
    Promise.resolve(options.subscriber.unsubscribe(channel)).catch(
      reportSubscriberError,
    );
  };

  return {
    async publish(event, payload) {
      instrumentation.record({
        type: "eventBus",
        eventName: event.name,
      });

      await options.publisher.publish(
        channelForEvent(channelPrefix, event.name),
        JSON.stringify({ eventName: event.name, payload }),
      );
    },

    subscribe<E extends DomainEventDef>(
      event: E,
      handler: (payload: InferEventPayload<E>) => void | Promise<void>,
    ) {
      let eventHandlers = handlers.get(event.name);
      if (!eventHandlers) {
        eventHandlers = new Set();
        handlers.set(event.name, eventHandlers);
      }

      eventHandlers.add(handler as (payload: unknown) => void | Promise<void>);
      ensureSubscribed(event.name);

      return () => {
        const currentHandlers = handlers.get(event.name);
        if (!currentHandlers) return;
        currentHandlers.delete(
          handler as (payload: unknown) => void | Promise<void>,
        );
        if (currentHandlers.size === 0) {
          handlers.delete(event.name);
          unsubscribeChannel(event.name);
        }
      };
    },
  };
}

export function createRedisEventBusProvider(
  options: CreateRedisEventBusProviderOptions = {},
): RedisEventBusProvider {
  const providerMetadata = {
    packageName: "@beignet/provider-event-bus-redis",
    ports: ["eventBus", "redisEventBus"],
    env: [
      "REDIS_EVENT_BUS_URL",
      "REDIS_EVENT_BUS_DB",
      "REDIS_EVENT_BUS_CHANNEL_PREFIX",
      "REDIS_EVENT_BUS_CONNECT_TIMEOUT_MS",
      "REDIS_EVENT_BUS_MAX_RETRIES_PER_REQUEST",
      "REDIS_EVENT_BUS_CONNECT_MAX_ATTEMPTS",
    ],
    watchers: ["eventBus"],
  } as const;

  return createProvider({
    name: "event-bus-redis",
    metadata: providerMetadata,

    config: {
      schema: RedisEventBusConfigSchema,
      envPrefix: "REDIS_EVENT_BUS_",
      // Defined options win over env-derived values at config load time.
      overrides: {
        DB: options.db,
        CHANNEL_PREFIX: options.channelPrefix,
        CONNECT_TIMEOUT_MS: options.connectTimeoutMs,
        MAX_RETRIES_PER_REQUEST: options.maxRetriesPerRequest,
      },
    },

    async setup({ config, ports }) {
      if (!config) {
        throw new Error(
          "[createRedisEventBusProvider] Missing Redis event bus config. " +
            "Please set REDIS_EVENT_BUS_URL.",
        );
      }

      const configuredChannelPrefix =
        config.CHANNEL_PREFIX ??
        options.channelPrefix ??
        DEFAULT_CHANNEL_PREFIX;
      const connectMaxAttempts =
        config.CONNECT_MAX_ATTEMPTS ?? DEFAULT_CONNECT_MAX_ATTEMPTS;
      const publisherRetry = createRetryStrategy(connectMaxAttempts);
      const subscriberRetry = createRetryStrategy(connectMaxAttempts);
      const redisOptions = {
        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,
      };

      const publisher = new Redis(config.URL, {
        ...redisOptions,
        retryStrategy: options.retryStrategy ?? publisherRetry.strategy,
      });
      const subscriber = new Redis(config.URL, {
        ...redisOptions,
        retryStrategy: options.retryStrategy ?? subscriberRetry.strategy,
      });

      try {
        await publisher.connect();
        publisherRetry.markConnected();
        await subscriber.connect();
        subscriberRetry.markConnected();
      } catch (error) {
        publisher.disconnect();
        subscriber.disconnect();
        throw new Error(
          `[createRedisEventBusProvider] Failed to connect to Redis at ${redactConnectionUrl(
            config.URL,
          )}: ${errorMessage(error)}`,
        );
      }

      const eventBus = createRedisEventBus({
        publisher,
        subscriber,
        channelPrefix: configuredChannelPrefix,
        instrumentation: ports,
      });

      return {
        ports: {
          eventBus,
          redisEventBus: {
            publisher,
            subscriber,
            channelPrefix: configuredChannelPrefix,
            checkHealth: () =>
              checkRedisEventBusHealth(
                publisher,
                subscriber,
                configuredChannelPrefix,
              ),
          },
        } satisfies RedisEventBusProviderPorts,
        async stop() {
          try {
            await subscriber.quit();
          } catch {
            subscriber.disconnect();
          }
          try {
            await publisher.quit();
          } catch {
            publisher.disconnect();
          }
        },
      };
    },
  });
}