import type { Handler, PubSubDriver } from "./index.ts";

export interface NatsClient {
  publish(subject: string, data: Uint8Array): void | Promise<void>;
  subscribe(subject: string, handler: (data: Uint8Array, subject: string) => void): () => void;
  close?(): void | Promise<void>;
}

export function natsDriver(client: NatsClient): PubSubDriver {
  const encoder = new TextEncoder();
  const decoder = new TextDecoder();
  return {
    publish(topic, message) {
      return client.publish(topic, encoder.encode(JSON.stringify(message)));
    },
    subscribe(pattern, handler) {
      const subject =
        pattern === "*" ? ">" : pattern.endsWith(":*") ? `${pattern.slice(0, -2)}:>` : pattern;
      return client.subscribe(subject, (data, topic) => {
        const raw = decoder.decode(data);
        let value: unknown = raw;
        try {
          value = JSON.parse(raw);
        } catch {
          /* raw broker payload */
        }
        void handler(value, topic);
      });
    },
    close: () => client.close?.(),
  };
}

export interface KafkaClient {
  publish(topic: string, value: string): void | Promise<void>;
  subscribe(pattern: string, handler: (value: string, topic: string) => void): () => void;
  close?(): void | Promise<void>;
}

/** Kafka adapter contract; consumer-group/rebalance policy remains owned by the selected client. */
export function kafkaDriver(client: KafkaClient): PubSubDriver {
  return {
    publish(topic, message) {
      return client.publish(topic, JSON.stringify(message));
    },
    subscribe(pattern, handler: Handler) {
      return client.subscribe(pattern, (raw, topic) => {
        let value: unknown = raw;
        try {
          value = JSON.parse(raw);
        } catch {
          /* raw broker payload */
        }
        void handler(value, topic);
      });
    },
    close: () => client.close?.(),
  };
}