import type { TagCache } from "./memory.ts";

export interface CacheInvalidationBus {
  publish(topic: string, message: unknown): void | Promise<void>;
  subscribe(pattern: string, handler: (message: unknown) => void | Promise<void>): () => void;
}

export interface DistributedInvalidationOptions {
  namespace?: string;
  instanceId?: string;
  onError?: (error: unknown) => void;
}

type InvalidationMessage =
  | { source: string; operation: "tags"; tags: string[] }
  | { source: string; operation: "key"; key: string }
  | { source: string; operation: "clear" };

export interface DistributedInvalidation {
  invalidateTag(tag: string): Promise<number>;
  invalidateTags(tags: Iterable<string>): Promise<number>;
  delete(key: string): Promise<boolean>;
  clear(): Promise<void>;
  close(): void;
}

/**
 * Propagate cache invalidations over any structurally compatible pub/sub bus.
 * The bus is intentionally not closed because applications commonly share it.
 */
export function connectCacheInvalidation<V>(
  cache: TagCache<V>,
  bus: CacheInvalidationBus,
  options: DistributedInvalidationOptions = {},
): DistributedInvalidation {
  const namespace = options.namespace?.trim() || "default";
  if (!/^[a-zA-Z0-9._-]+$/.test(namespace)) {
    throw new TypeError("cache invalidation namespace contains unsupported characters");
  }
  const source = options.instanceId?.trim() || crypto.randomUUID();
  const topic = `wrnexus:cache:${namespace}:invalidate`;
  let closed = false;

  const publish = async (message: InvalidationMessage): Promise<void> => {
    if (closed) throw new Error("WRN-CACHE-INVALIDATION-CLOSED: invalidation channel is closed");
    try {
      await bus.publish(topic, message);
    } catch (error) {
      options.onError?.(error);
      throw error;
    }
  };

  const unsubscribe = bus.subscribe(topic, (value) => {
    if (!value || typeof value !== "object") return;
    const message = value as Partial<InvalidationMessage>;
    if (message.source === source) return;
    if (message.operation === "tags" && Array.isArray(message.tags)) {
      cache.invalidateTags(message.tags.filter((tag): tag is string => typeof tag === "string"));
    } else if (message.operation === "key" && typeof message.key === "string") {
      cache.delete(message.key);
    } else if (message.operation === "clear") {
      cache.clear();
    }
  });

  const invalidateTags = async (input: Iterable<string>): Promise<number> => {
    const tags = [...new Set(input)].filter(Boolean);
    const removed = cache.invalidateTags(tags);
    await publish({ source, operation: "tags", tags });
    return removed;
  };

  return {
    async invalidateTag(tag) {
      return invalidateTags([tag]);
    },
    invalidateTags,
    async delete(key) {
      const removed = cache.delete(key);
      await publish({ source, operation: "key", key });
      return removed;
    },
    async clear() {
      cache.clear();
      await publish({ source, operation: "clear" });
    },
    close() {
      if (closed) return;
      closed = true;
      unsubscribe();
    },
  };
}