// Stream 操作
import { getClient } from './connection'

export async function streamInfo(id: string, key: string): Promise<any> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return client.xinfo('STREAM', key)
}

export async function streamRange(
  id: string, key: string, start: string, end: string, count: number
): Promise<any[]> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return client.xrange(key, start, end, 'COUNT', count)
}

export async function streamAdd(
  id: string, key: string, streamId: string, fieldValues: string[]
): Promise<string | null> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return client.xadd(key, streamId, ...fieldValues)
}

export async function streamTrim(id: string, key: string, maxLen: number): Promise<number> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return client.xtrim(key, 'MAXLEN', maxLen)
}

export async function streamDel(id: string, key: string, ...ids: string[]): Promise<number> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return client.xdel(key, ...ids)
}

// Consumer group operations
export async function streamGroups(id: string, key: string): Promise<any[]> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return client.xinfo('GROUPS', key) as Promise<any[]>
}

export async function streamConsumers(id: string, key: string, group: string): Promise<any[]> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return client.xinfo('CONSUMERS', key, group) as Promise<any[]>
}

export async function streamGroupCreate(
  id: string, key: string, group: string, startId: string, mkstream: boolean
): Promise<string> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  const args: any[] = [key, group, startId]
  if (mkstream) args.push('MKSTREAM')
  return (client as any).xgroup('CREATE', ...args)
}

export async function streamGroupDestroy(id: string, key: string, group: string): Promise<number> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return (client as any).xgroup('DESTROY', key, group)
}

export async function streamAck(id: string, key: string, group: string, ...ids: string[]): Promise<number> {
  const client = getClient(id)
  if (!client) throw new Error(`Client ${id} not found`)
  return (client as any).xack(key, group, ...ids)
}