// Acknowledge and delete a single entry
const result = await redis.xackdel("mystream", "mygroup", "1638360173533-0");
console.log(result); // Array of results for each ID

// Acknowledge and delete multiple entries
const result = await redis.xackdel(
  "mystream", 
  "mygroup",
  "1638360173533-0",
  "1638360173533-1",
  "1638360173533-2"
);
console.log(result); // Array of results for each ID

// Keep consumer group references when deleting entries
// Useful when you want to delete entries but maintain group tracking
const result = await redis.xackdel(
  "mystream",
  "mygroup",
  "KEEPREF",
  "1638360173533-0",
  "1638360173533-1",
);
console.log(result); // Array of results for each ID

// Delete consumer group references along with entries
// Useful for complete cleanup of processed messages
const result = await redis.xackdel(
  "mystream",
  "mygroup",
  "DELREF",
  "1638360173533-0",
  "1638360173533-1",
);
console.log(result); // Array of results for each ID

// Only acknowledge messages without deleting them
// Useful when you want to mark as processed but keep for audit/debugging
const result = await redis.xackdel(
  "mystream",
  "mygroup",
  "ACKED",
  "1638360173533-0",
);
console.log(result); // Array of results for each ID
// Entry remains in stream but is acknowledged

// Create a consumer group
await redis.xgroup("CREATE", "orders", "processors", "0", "MKSTREAM");

// Add some orders to the stream
await redis.xadd("orders", "*", { order_id: "123", status: "pending" });
await redis.xadd("orders", "*", { order_id: "124", status: "pending" });

// Read messages as a consumer
const messages = await redis.xreadgroup(
  "GROUP", "processors", "consumer1",
  "COUNT", 2,
  "STREAMS", "orders", ">"
);

// Process messages and acknowledge + delete them
for (const [stream, entries] of messages) {
  for (const [id, fields] of entries) {
    // Process the order...
    console.log(`Processing order ${fields.order_id}`);
    
    // Acknowledge and delete the entry (default behavior)
    await redis.xackdel("orders", "processors", id);
    
    // Or use DELREF for complete cleanup
    // await redis.xackdel("orders", "processors", "DELREF", id);
  }
}

// Read pending messages
const pending = await redis.xreadgroup(
  "GROUP", "tasks", "worker1",
  "COUNT", 10,
  "STREAMS", "task-queue", ">"
);

if (pending && pending.length > 0) {
  const [stream, entries] = pending[0];
  const processedIds: string[] = [];
  
  // Process each task
  for (const [id, fields] of entries) {
    try {
      // Process task...
      console.log(`Processing task ${id}`);
      processedIds.push(id);
    } catch (error) {
      console.error(`Failed to process ${id}:`, error);
    }
  }
  
  // Acknowledge and delete all successfully processed tasks
  // Using DELREF to completely remove references
  if (processedIds.length > 0) {
    const result = await redis.xackdel("task-queue", "tasks", "DELREF", ...processedIds);
    console.log(`Results: ${result}`); // Array of results for each ID
  }
}

// Process messages but keep them for audit/debugging purposes
const messages = await redis.xreadgroup(
  "GROUP", "audit", "auditor1",
  "COUNT", 5,
  "STREAMS", "events", ">"
);

for (const [stream, entries] of messages) {
  const entryIds = entries.map(([id]) => id);
  
  // Process and acknowledge but don't delete
  for (const [id, fields] of entries) {
    // Audit the event...
    console.log(`Auditing event ${JSON.stringify(fields)}`);
  }
  
  // Acknowledge without deleting (keep for audit trail)
  const result = await redis.xackdel("events", "audit", "ACKED", ...entryIds);
  console.log(`Acknowledged ${result.length} events (kept in stream)`);
}