function normalizeUrl(value) {
  const raw = String(value ?? '').trim();
  if (!raw) return null;
  try {
    const url = new URL(raw);
    if (!['redis:', 'rediss:'].includes(url.protocol)) {
      throw new Error(`Unsupported protocol: ${url.protocol}`);
    }
    return raw;
  } catch (err) {
    throw new Error(`Invalid MEMORY_REDIS_STREAMS_URL: ${err instanceof Error ? err.message : err}`);
  }
}

function normalizeStream(value) {
  const raw = String(value ?? 'memind:memory-events').trim();
  if (!/^[a-zA-Z0-9:_-]+$/.test(raw)) {
    throw new Error(`Invalid MEMORY_REDIS_STREAMS_STREAM: ${raw}`);
  }
  return raw;
}

function flattenFields(input = {}) {
  return {
    user_id: String(input.userId ?? ''),
    session_id: String(input.sessionId ?? ''),
    event_type: String(input.eventType ?? 'memory.write'),
    messages_json: JSON.stringify(Array.isArray(input.messages) ? input.messages : []),
    created_at: new Date().toISOString(),
  };
}

export async function createRedisStreamsClient({
  url,
  stream = 'memind:memory-events',
  importRedis = (specifier) => import(specifier),
} = {}) {
  const resolvedUrl = normalizeUrl(url);
  const resolvedStream = normalizeStream(stream);
  if (!resolvedUrl) throw new Error('createRedisStreamsClient requires MEMORY_REDIS_STREAMS_URL');
  const imported = await importRedis('redis');
  const createClient = imported?.createClient ?? imported?.default?.createClient;
  if (typeof createClient !== 'function') throw new Error('redis module does not export createClient');
  const client = createClient({ url: resolvedUrl });
  let connected = false;

  async function ensureConnected() {
    if (connected) return;
    await client.connect();
    connected = true;
  }

  return {
    async write(input = {}) {
      await ensureConnected();
      await client.xAdd(resolvedStream, '*', flattenFields(input));
      return { saved: 1, analyzed: 0, memories: 0 };
    },
    async close() {
      if (!connected) return;
      connected = false;
      await client.quit?.();
    },
  };
}

export function createRedisStreamsMemoryBackend({
  enabled = false,
  url = null,
  stream = 'memind:memory-events',
  client = null,
  unavailableReason = null,
} = {}) {
  const resolvedUrl = normalizeUrl(url);
  const resolvedStream = normalizeStream(stream);
  const configured = Boolean(enabled && resolvedUrl);
  const hasClient = Boolean(client?.write);
  const wired = Boolean(configured && hasClient);
  const reason = unavailableReason
    ?? (enabled
      ? (!resolvedUrl ? 'url_not_configured' : 'client_not_configured')
      : 'not_configured');

  return {
    name: 'redis-streams',
    category: 'behavior',
    role: 'event-tracking',
    flag: 'MEMORY_REDIS_STREAMS_ENABLED',
    unavailableReason: reason,
    url: resolvedUrl,
    stream: resolvedStream,

    isAvailable() {
      return Boolean(wired);
    },

    getUnavailableReason() {
      return this.isAvailable() ? null : reason;
    },

    async write(input = {}) {
      if (!this.isAvailable()) return { saved: 0, analyzed: 0, memories: 0 };
      const result = await client.write({ ...input, url: resolvedUrl, stream: resolvedStream });
      return {
        saved: Number(result?.saved ?? 0),
        analyzed: Number(result?.analyzed ?? 0),
        memories: Number(result?.memories ?? 0),
      };
    },
  };
}