### Starting the Shard Manager

Source: https://context7.com/devsisters/shardcake/llms.txt

This section details how to run the Shard Manager process, which is responsible for assigning shards to pods. It outlines the required layers and provides a basic setup using in-memory storage and local health checks.

```APIDOC
## Starting the Shard Manager (`Server.run`)

The Shard Manager is an independent process that assigns shards to pods. Run it once. It requires `Storage`, `Pods`, `PodsHealth`, and `ShardManager` layers. The simplest setup uses in-memory storage and local (ping-based) health checks.

```scala
import com.devsisters.shardcake._
import com.devsisters.shardcake.interfaces._
import zio._

object ShardManagerApp extends ZIOAppDefault {
  def run: Task[Nothing] = 
    Server.run.provide(
      ZLayer.succeed(ManagerConfig.default), // apiPort=8080, numberOfShards=300, rebalanceInterval=20s
      ZLayer.succeed(GrpcConfig.default),
      PodsHealth.local,   // ping-based health check (dev/single-pod)
      GrpcPods.live,      // gRPC messaging protocol
      Storage.memory,     // in-memory storage (use StorageRedis.live for production)
      ShardManager.live
    )
}
// Default ManagerConfig values:
// numberOfShards = 300, apiPort = 8080, rebalanceInterval = 20 seconds
// pingTimeout = 3 seconds, rebalanceRate = 2%
```
```

--------------------------------

### Start Shard Manager with ZIO Layers

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

This ZIO application starts the Shard Manager with default configurations for manager, gRPC, pods health, pods communication, storage, and the shard manager itself. It uses ZLayer to provide these dependencies.

```scala
import com.devsisters.shardcake._
import com.devsisters.shardcake.interfaces._
import zio._

object ShardManagerApp extends ZIOAppDefault {
  def run: Task[Nothing] = 
    Server.run.provide(
      ZLayer.succeed(ManagerConfig.default),
      ZLayer.succeed(GrpcConfig.default),
      PodsHealth.local, // just ping a pod to see if it's alive
      GrpcPods.live,    // use gRPC protocol
      Storage.memory,   // store data in memory
      ShardManager.live // shard manager logic
    )
}
```

--------------------------------

### Configure Redis Storage Layer

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

This example shows how to build the Redis storage layer. It requires a RedisConfig with keys for assignments and pods, and a Redis object providing connection and pub/sub capabilities.

```scala
import com.devsisters.shardcake.StorageRedis.Redis
import dev.profunktor.redis4cats.Redis
import dev.profunktor.redis4cats.connection.RedisClient
import dev.profunktor.redis4cats.data.RedisCodec
import dev.profunktor.redis4cats.effect.Log
import dev.profunktor.redis4cats.pubsub.PubSub
import zio.interop.catz._
import zio.{ Task, ZEnvironment, ZIO, ZLayer }

val redis: ZLayer[Any, Throwable, Redis] =
  ZLayer.scopedEnvironment {
    implicit val runtime: zio.Runtime[Any] = zio.Runtime.default
    implicit val logger: Log[Task]         = new Log[Task] {
      override def debug(msg: => String): Task[Unit] = ZIO.unit
      override def error(msg: => String): Task[Unit] = ZIO.logError(msg)
      override def info(msg: => String): Task[Unit]  = ZIO.logDebug(msg)
    }

    (for {
      client   <- RedisClient[Task].from("redis://foobared@localhost")
      commands <- Redis[Task].fromClient(client, RedisCodec.Utf8)
      pubSub   <- PubSub.mkPubSubConnection[Task, String, String](client, RedisCodec.Utf8)
    } yield ZEnvironment(commands, pubSub)).toScopedZIO
  }
```

--------------------------------

### Start Shard Manager Server.run

Source: https://context7.com/devsisters/shardcake/llms.txt

Runs the Shard Manager process, which assigns shards to pods. Requires Storage, Pods, PodsHealth, and ShardManager layers. Uses in-memory storage and local health checks by default.

```scala
import com.devsisters.shardcake._
import com.devsisters.shardcake.interfaces._
import zio._

object ShardManagerApp extends ZIOAppDefault {
  def run: Task[Nothing] = 
    Server.run.provide(
      ZLayer.succeed(ManagerConfig.default), // apiPort=8080, numberOfShards=300, rebalanceInterval=20s
      ZLayer.succeed(GrpcConfig.default),
      PodsHealth.local,   // ping-based health check (dev/single-pod)
      GrpcPods.live,      // gRPC messaging protocol
      Storage.memory,     // in-memory storage (use StorageRedis.live for production)
      ShardManager.live
    )
}
// Default ManagerConfig values:
// numberOfShards = 300, apiPort = 8080, rebalanceInterval = 20 seconds
// pingTimeout = 3 seconds, rebalanceRate = 2%
```

--------------------------------

### Register Entities and Send Messages

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Register entity behavior with the Sharding system and use the messenger to send messages to entities. This example demonstrates sending multiple Join messages to a guild.

```scala
  val program =
    for {
      _     <- Sharding.registerEntity(Guild, behavior)
      _     <- Sharding.registerScoped
      guild <- Sharding.messenger(Guild)
      _     <- guild.send("guild1")(Join("user1", _)).debug
      _     <- guild.send("guild1")(Join("user2", _)).debug
      _     <- guild.send("guild1")(Join("user3", _)).debug
      _     <- guild.send("guild1")(Join("user4", _)).debug
      _     <- guild.send("guild1")(Join("user5", _)).debug
      _     <- guild.send("guild1")(Join("user6", _)).debug
    } yield ()
```

--------------------------------

### ZIO Metrics Integration

Source: https://context7.com/devsisters/shardcake/llms.txt

Shardcake exposes metrics via ZIO Metrics, which can be integrated with various backends like Prometheus and Datadog using `zio-metrics-connectors`. The example shows how to wire the Prometheus connector into your application.

```APIDOC
## ZIO Metrics Integration

Shardcake exposes metrics via ZIO Metrics that can be connected to any backend (Prometheus, Datadog, etc.) using `zio-metrics-connectors`.

```scala
// Shard Manager metrics (gauges and counters exposed automatically):
// shardcake.pods               - gauge: number of registered pods
// shardcake.shards_assigned    - gauge: number of shards assigned to a pod
// shardcake.shards_unassigned  - gauge: number of unassigned shards
// shardcake.rebalances         - counter: total rebalance operations
// shardcake.pod_health_checked - counter: total pod health checks

// Pod-level metrics:
// shardcake.shards             - gauge: shards on this pod
// shardcake.entities           - gauge: live entities on this pod
// shardcake.singletons         - gauge: active singletons on this pod

// Example: wire Prometheus connector
import zio.metrics.connectors.prometheus._

myApp.provide(
  prometheusLayer,          // zio-metrics-connectors-prometheus
  Sharding.live,
  // ... other layers
)
```
```

--------------------------------

### Send Message to Entity by ID

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Example of sending a message to a guild entity using its ID, demonstrating location transparency. Requires ZIO context for current user.

```scala
def joinGuild(guildId: GuildId): ZIO[Context, Throwable, GuildState] = 
  for {
    userId     <- getCurrentUserFromContext
    guildState <- guild.send(guildId)(JoinGuild(userId, _))
  } yield guildState
```

--------------------------------

### Get Messenger to Send Messages

Source: https://context7.com/devsisters/shardcake/llms.txt

Obtains a Messenger for an entity type, handling transparent routing to the correct pod. Retries on PodUnavailable or EntityNotManagedByThisPod errors.

```scala
import com.devsisters.shardcake.{ Sharding, Messenger }
import zio._
import scala.util.Try

val program = 
  for {
    _     <- Sharding.registerEntity(Guild, behavior)
    _     <- Sharding.registerScoped
    guild <- Sharding.messenger(Guild) // Messenger[GuildMessage]

    // send with reply (blocks until response)
    res1  <- guild.send("guild1")(GuildMessage.Join("alice", _))
    _     <- ZIO.debug(res1) // Success(Set(alice))

    // send without waiting for reply (fire and forget)
    _     <- guild.sendDiscard("guild1")(GuildMessage.Leave("alice"))

    // send with custom timeout override
    guildNoTimeout <- Sharding.messenger(Guild, Messenger.MessengerTimeout.NoTimeout)
    res2           <- guildNoTimeout.send("guild2")(GuildMessage.Join("bob", _))
  } yield ()
```

--------------------------------

### Define Storage Trait

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

The Storage trait defines methods for storing and accessing pod and shard assignments. It includes functions for getting and saving assignments, getting and saving pods, and a stream for assignment updates.

```scala
trait Storage {
  def getAssignments: Task[Map[ShardId, Option[PodAddress]]]
  def saveAssignments(assignments: Map[ShardId, Option[PodAddress]]): Task[Unit]
  
  def assignmentsStream: ZStream[Any, Throwable, Map[Int, Option[PodAddress]]]
  
  def getPods: Task[Map[PodAddress, Pod]]
  def savePods(pods: Map[PodAddress, Pod]): Task[Unit]
}
```

--------------------------------

### Getting a Messenger to Send Messages

Source: https://context7.com/devsisters/shardcake/llms.txt

This function obtains a Messenger for a specific entity type, enabling transparent routing of messages to the correct pod and entity. It handles retries for unavailable pods or entities.

```APIDOC
## `Sharding.messenger` — Get a Messenger to Send Messages to Entities

Obtain a `Messenger[Msg]` for an entity type. The messenger handles routing transparently: it looks up which pod hosts the target entity and sends the message there, retrying on `PodUnavailable` or `EntityNotManagedByThisPod` errors.

```scala
import com.devsisters.shardcake.{ Sharding, Messenger }
import zio._
import scala.util.Try

val program = 
  for {
    _     <- Sharding.registerEntity(Guild, behavior)
    _     <- Sharding.registerScoped
    guild <- Sharding.messenger(Guild) // Messenger[GuildMessage]

    // send with reply (blocks until response)
    res1  <- guild.send("guild1")(GuildMessage.Join("alice", _))
    _     <- ZIO.debug(res1) // Success(Set(alice))

    // send without waiting for reply (fire and forget)
    _     <- guild.sendDiscard("guild1")(GuildMessage.Leave("alice"))

    // send with custom timeout override
    guildNoTimeout <- Sharding.messenger(Guild, Messenger.MessengerTimeout.NoTimeout)
    res2           <- guildNoTimeout.send("guild2")(GuildMessage.Join("bob", _))
  } yield () 
```
```

--------------------------------

### Provide Dependencies for Application

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Configure and provide the necessary dependencies for running the Shardcake application, including configuration, serialization, storage, and Sharding services.

```scala
def run: Task[Unit] =
  ZIO.scoped(program).provide(
    ZLayer.succeed(Config.default),
    ZLayer.succeed(GrpcConfig.default),
    Serialization.javaSerialization, // use java serialization for messages
    Storage.memory,                  // store data in memory
    ShardManagerClient.liveWithSttp, // client to communicate with the Shard Manager
    GrpcPods.live,                   // use gRPC protocol
    GrpcShardingService.live,        // expose gRPC service
    Sharding.live                    // sharding logic
  )
```

--------------------------------

### Full Pod Application Wiring with ZIO

Source: https://context7.com/devsisters/shardcake/llms.txt

Registers entities, connects to the Shard Manager, and provides required ZIO layers. Uses ZIO.scoped for graceful unregistration. Ensure all necessary ZIO layers are provided.

```scala
import com.devsisters.shardcake._
import com.devsisters.shardcake.interfaces._
import zio._

object GuildApp extends ZIOAppDefault {
  val program: RIO[Sharding with Scope, Unit] = 
    for {
      _     <- Sharding.registerEntity(
                 Guild,
                 GuildBehavior.behavior,
                 terminateMessage = p => Some(GuildMessage.Terminate(p))
               )
      _     <- Sharding.registerScoped
      guild <- Sharding.messenger(Guild)
      // Send 6 join requests; the 6th will fail since max is 5
      _     <- ZIO.foreach(1 to 6)(i => guild.send("guild1")(GuildMessage.Join(s"user$i", _)).debug)
    } yield () 

  def run: Task[Unit] = 
    ZIO.scoped(program).provide(
      ZLayer.succeed(Config.default),          // selfHost=localhost, shardingPort=54321
      ZLayer.succeed(GrpcConfig.default),
      Serialization.javaSerialization,         // or KryoSerialization.live for production
      Storage.memory,                          // or StorageRedis.live for production
      ShardManagerClient.liveWithSttp,         // GraphQL client to Shard Manager
      GrpcPods.live,                           // gRPC pod-to-pod protocol
      GrpcShardingService.live,                // expose gRPC server on this pod
      Sharding.live
    )
}
```

--------------------------------

### Run a Background Process on One Pod Only

Source: https://context7.com/devsisters/shardcake/llms.txt

Register a background process using `Sharding.registerSingleton` that runs on exactly one pod at a time. If the pod hosting the singleton fails, it is automatically restarted on another pod.

```scala
import com.devsisters.shardcake.Sharding
import zio._

// A background cleanup job that should run on only one pod
val cleanupJob: UIO[Nothing] = 
  (ZIO.debug("Running cleanup...") *> ZIO.sleep(1.minute)).forever

Sharding.registerSingleton("cleanup-job", cleanupJob)
```

--------------------------------

### Wire Prometheus Connector with ZIO Metrics

Source: https://context7.com/devsisters/shardcake/llms.txt

This snippet shows how to provide the Prometheus layer to your application, enabling ZIO Metrics to report Shardcake metrics to a Prometheus backend. Ensure the `zio-metrics-connectors-prometheus` dependency is included.

```scala
import zio.metrics.connectors.prometheus._

myApp.provide(
  prometheusLayer,
  Sharding.live,
  // ... other layers
)
```

--------------------------------

### Messenger.sendAndReceiveStreamAutoRestart

Source: https://context7.com/devsisters/shardcake/llms.txt

Provides resilient streaming by automatically restarting the stream after a rebalance. It uses a cursor to resume from the last received position, ensuring no data is lost during interruptions.

```APIDOC
## `Messenger.sendAndReceiveStreamAutoRestart` — Resilient Streaming with Cursor

Automatically restarts the stream after a rebalance using a cursor to resume from the last received position. The cursor is updated with each response element and re-sent to the entity if the stream is interrupted by a pod rebalance.

```scala
import com.devsisters.shardcake.StreamReplier

// Message type that carries a cursor (offset) and a stream replier
case class WatchEvents(fromOffset: Long, replier: StreamReplier[GameEvent]) extends GuildMessage

// guild: Messenger[GuildMessage]
val resilientStream: ZStream[Any, Throwable, GameEvent] = 
  guild.sendAndReceiveStreamAutoRestart(entityId = "guild42", cursor = 0L)(
    msg = (offset, replier) => WatchEvents(offset, replier),
    updateCursor = (currentOffset, event) => event.offset
  )

resilientStream
  .foreach(event => ZIO.debug(s"Event at offset ${event.offset}: ${event.data}"))
// If the pod hosting guild42 is rebalanced, the stream automatically restarts
// from the last seen offset after a 200ms delay.
```
```

--------------------------------

### Sharding.registerSingleton

Source: https://context7.com/devsisters/shardcake/llms.txt

Allows running a background process on exactly one pod at a time. This is useful for tasks that should not be duplicated across multiple instances, ensuring only one pod executes the process.

```APIDOC
## `Sharding.registerSingleton` — Run a Background Process on One Pod Only

Register a named background process that runs on exactly one pod at a time (the pod hosting shard 1). If that pod goes down, the singleton is automatically restarted on another pod. No messages can be sent to singletons.

```scala
import com.devsisters.shardcake.Sharding
import zio._

// A background cleanup job that should run on only one pod
val cleanupJob: UIO[Nothing] = 
  (ZIO.debug("Running cleanup...") *> ZIO.sleep(1.minute)).forever

Sharding.registerSingleton("cleanup-job", cleanupJob)
// Only the pod assigned shard #1 will run cleanupJob.
// If that pod is replaced, the new shard-1 holder starts the job automatically.
```
```

--------------------------------

### Add Shardcake Manager and gRPC Protocol Dependencies

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Include these library dependencies in your build.sbt file to use the Shard Manager and its gRPC protocol.

```scala
libraryDependencies += "com.devsisters" %% "shardcake-manager"       % "2.7.1"
libraryDependencies += "com.devsisters" %% "shardcake-protocol-grpc" % "2.7.1"
```

--------------------------------

### Resilient Streaming with Cursor and Auto-Restart

Source: https://context7.com/devsisters/shardcake/llms.txt

Implement resilient streaming using `sendAndReceiveStreamAutoRestart`. This method automatically restarts the stream after a rebalance, resuming from the last received position using a provided cursor.

```scala
import com.devsisters.shardcake.StreamReplier

// Message type that carries a cursor (offset) and a stream replier
case class WatchEvents(fromOffset: Long, replier: StreamReplier[GameEvent]) extends GuildMessage

// guild: Messenger[GuildMessage]
val resilientStream: ZStream[Any, Throwable, GameEvent] =
  guild.sendAndReceiveStreamAutoRestart(entityId = "guild42", cursor = 0L)(
    msg = (offset, replier) => WatchEvents(offset, replier),
    updateCursor = (currentOffset, event) => event.offset
  )

resilientStream
  .foreach(event => ZIO.debug(s"Event at offset ${event.offset}: ${event.data}"))
```

--------------------------------

### Broadcast Messages to All Pods

Source: https://context7.com/devsisters/shardcake/llms.txt

Use `Sharding.registerTopic` and `Sharding.broadcaster` to broadcast messages to all registered pods simultaneously. This is useful for sending commands or queries to every pod in the cluster.

```scala
import com.devsisters.shardcake.{ TopicType, Broadcaster, Replier, Sharding }
import zio._
import scala.util.Try

// Define topic type
object SystemTopic extends TopicType[SystemMessage]("system")

sealed trait SystemMessage
case class Ping(replier: Replier[String]) extends SystemMessage

// Register the topic handler on each pod
Sharding.registerTopic(
  SystemTopic,
  behavior = (topic: String, queue: Dequeue[SystemMessage]) =>
    queue.take.flatMap {
      case Ping(replier) => replier.reply(s"pong from $topic")
    }.forever
)

// Broadcast a message and collect responses from all pods
val results: UIO[Map[PodAddress, Try[String]]] =
  for {
    broadcaster <- Sharding.broadcaster(SystemTopic)
    responses   <- broadcaster.broadcast("global")(Ping(_))
    _           <- ZIO.foreachDiscard(responses) { case (pod, result) =>
                     ZIO.debug(s"$pod -> $result")
                   }
  } yield responses
```

--------------------------------

### Add Shardcake Entities and gRPC Protocol Dependencies

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Include these library dependencies in your build.sbt file to define and manage entity behavior using Shardcake.

```scala
libraryDependencies += "com.devsisters" %% "shardcake-entities"      % "2.7.1"
libraryDependencies += "com.devsisters" %% "shardcake-protocol-grpc" % "2.7.1"
```

--------------------------------

### Add Kryo Serialization Dependency

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

Include the Kryo serialization library as a dependency to enable its use in your project. This is a common choice for binary serialization.

```scala
libraryDependencies += "com.devsisters" %% "shardcake-serialization-kryo" % "2.7.1"
```

--------------------------------

### Sharding.registerTopic + Sharding.broadcaster

Source: https://context7.com/devsisters/shardcake/llms.txt

Enables broadcasting messages to all registered pods simultaneously. A topic is registered, and then a broadcaster is used to send messages and collect responses from all pods in parallel.

```APIDOC
## `Sharding.registerTopic` + `Sharding.broadcaster` — Broadcast to All Pods

Register a topic type that broadcasts messages to all registered pods simultaneously. Use `Sharding.broadcaster` to get a `Broadcaster[Msg]` and call `broadcast` to send to all pods in parallel, collecting per-pod results.

```scala
import com.devsisters.shardcake.{ TopicType, Broadcaster, Replier, Sharding }
import zio._
import scala.util.Try

// Define topic type
object SystemTopic extends TopicType[SystemMessage]("system")

sealed trait SystemMessage
case class Ping(replier: Replier[String]) extends SystemMessage

// Register the topic handler on each pod
Sharding.registerTopic(
  SystemTopic,
  behavior = (topic: String, queue: Dequeue[SystemMessage]) =>
    queue.take.flatMap {
      case Ping(replier) => replier.reply(s"pong from $topic")
    }.forever
)

// Broadcast a message and collect responses from all pods
val results: UIO[Map[PodAddress, Try[String]]] = 
  for {
    broadcaster <- Sharding.broadcaster(SystemTopic)
    responses   <- broadcaster.broadcast("global")(Ping(_))
    _           <- ZIO.foreachDiscard(responses) { case (pod, result) =>
                     ZIO.debug(s"$pod -> $result")
                   }
  } yield responses
// Output (3 pods):
// PodAddress(host1, 54321) -> Success(pong from global)
// PodAddress(host2, 54321) -> Success(pong from global)
// PodAddress(host3, 54321) -> Success(pong from global)
```
```

--------------------------------

### Custom Shardcake Config for Pods

Source: https://context7.com/devsisters/shardcake/llms.txt

Overrides the default Config for tuning sharding behavior. Key settings include `numberOfShards` (must match across all pods and Shard Manager) and `serverVersion` (for rolling updates).

```scala
import com.devsisters.shardcake.Config
import sttp.client4.UriContext
import zio._

val productionConfig: Config = Config(
  numberOfShards               = 1000,             // 10x max expected pods
  selfHost                     = "10.0.1.42",       // current pod's IP
  shardingPort                 = 54321,
  shardManagerUri              = uri"http://shard-manager:8080/api/graphql",
  serverVersion                = "2.1.0",           // increment on deploy for rolling updates
  entityMaxIdleTime            = 10.minutes,
  entityTerminationTimeout     = 5.seconds,
  sendTimeout                  = 15.seconds,
  refreshAssignmentsRetryInterval = 5.seconds,
  unhealthyPodReportInterval   = 5.seconds,
  simulateRemotePods           = false,
  unregisterRetrySchedule      = Schedule.spaced(3.second) && Schedule.recurs(10)
)

ZLayer.succeed(productionConfig)
```

--------------------------------

### Add Redis Storage Dependency

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

To use Redis for storage, add this dependency to your project. This enables the production-ready Redis implementation of the Storage trait.

```scala
libraryDependencies += "com.devsisters" %% "shardcake-storage-redis" % "2.7.1"
```

--------------------------------

### Redis Storage Backend Layer

Source: https://context7.com/devsisters/shardcake/llms.txt

Provides a ZLayer for Redis storage using redis4cats. This is recommended for production to ensure the Shard Manager survives restarts. Ensure Redis is accessible at the configured URI.

```scala
import com.devsisters.shardcake.StorageRedis
import com.devsisters.shardcake.StorageRedis.Redis
import dev.profunktor.redis4cats.{ Redis => R4C }
import dev.profunktor.redis4cats.connection.RedisClient
import dev.profunktor.redis4cats.data.RedisCodec
import dev.profunktor.redis4cats.effect.Log
import dev.profunktor.redis4cats.pubsub.PubSub
import zio.interop.catz._
import zio.{ Task, ZEnvironment, ZIO, ZLayer }

val redisLayer: ZLayer[Any, Throwable, Redis] = 
  ZLayer.scopedEnvironment {
    implicit val runtime: zio.Runtime[Any] = zio.Runtime.default
    implicit val logger: Log[Task] = new Log[Task] {
      def debug(msg: => String): Task[Unit] = ZIO.unit
      def error(msg: => String): Task[Unit] = ZIO.logError(msg)
      def info(msg: => String):  Task[Unit] = ZIO.logDebug(msg)
    }
    (for {
      client   <- RedisClient[Task].from("redis://foobared@localhost")
      commands <- R4C[Task].fromClient(client, RedisCodec.Utf8)
      pubSub   <- PubSub.mkPubSubConnection[Task, String, String](client, RedisCodec.Utf8)
    } yield ZEnvironment(commands, pubSub)).toScopedZIO
  }

// Use in Shard Manager or pod layer stack:
Server.run.provide(
  ZLayer.succeed(ManagerConfig.default),
  ZLayer.succeed(GrpcConfig.default),
  PodsHealth.local,
  GrpcPods.live,
  ZLayer.succeed(RedisConfig.default), // assignmentsKey="shard_assignments", podsKey="pods"
  redisLayer,
  StorageRedis.live,  // replaces Storage.memory
  ShardManager.live
)
```

--------------------------------

### Registering Pod Lifecycle

Source: https://context7.com/devsisters/shardcake/llms.txt

This function notifies the Shard Manager that a pod is ready to accept shard assignments. It automatically handles unregistration upon scope closure, triggering a rebalance.

```APIDOC
## `Sharding.registerScoped` — Register/Unregister Pod Lifecycle

Notify the Shard Manager that this pod is ready to accept shard assignments. `registerScoped` automatically calls `unregister` when the `Scope` closes (e.g., on graceful shutdown), triggering an immediate rebalance so shards are moved to other pods.

```scala
import com.devsisters.shardcake.Sharding
import zio._

val program: RIO[Sharding with Scope, Unit] = 
  for {
    _ <- Sharding.registerEntity(Guild, behavior)
    _ <- Sharding.registerScoped  // registers on start, unregisters on scope close
    // ... rest of application
  } yield () 
```
```

--------------------------------

### SBT Dependencies for Shardcake

Source: https://context7.com/devsisters/shardcake/llms.txt

Add these dependencies to your build.sbt file to include Shardcake modules. The core entity module with the gRPC protocol is common for production.

```scala
// Core entity sharding on pods
libraryDependencies += "com.devsisters" %% "shardcake-entities"           % "2.7.1"
// Shard Manager server
libraryDependencies += "com.devsisters" %% "shardcake-manager"            % "2.7.1"
// gRPC protocol for pod-to-pod communication
libraryDependencies += "com.devsisters" %% "shardcake-protocol-grpc"      % "2.7.1"
// Redis storage backend
libraryDependencies += "com.devsisters" %% "shardcake-storage-redis"      % "2.7.1"
// Kryo binary serialization
libraryDependencies += "com.devsisters" %% "shardcake-serialization-kryo" % "2.7.1"
// Kubernetes pod health checks
libraryDependencies += "com.devsisters" %% "shardcake-health-k8s"         % "2.7.1"
```

--------------------------------

### Send Message and Receive Stream of Replies

Source: https://context7.com/devsisters/shardcake/llms.txt

Use `sendAndReceiveStream` to send a single message and receive a `ZStream` of responses. This is suitable for scenarios like subscriptions or paginated results where multiple replies are expected.

```scala
import com.devsisters.shardcake.StreamReplier
import zio.stream.ZStream

// guild: Messenger[GuildMessage]
val listMembers: ZIO[Any, Throwable, Unit] =
  guild
    .sendAndReceiveStream("guild42")(GuildMessage.GetMembers(_))
    .flatMap(stream => stream.foreach(member => ZIO.debug(s"Member: $member")))

// In the entity behavior, reply with a stream:
// case GuildMessage.GetMembers(replier) =>
//   state.get.flatMap(members => replier.replyStream(ZStream.fromIterable(members)))
```

--------------------------------

### Sending a Message and Awaiting a Single Reply

Source: https://context7.com/devsisters/shardcake/llms.txt

This method sends a message to an entity and waits for a single response. The message constructor includes a `Replier` that must be used to send the reply. The call will time out if no reply is received within the configured duration.

```APIDOC
## `Messenger.send` — Send a Message and Await a Single Reply

Send a message that expects exactly one response. The message constructor receives a `Replier[Res]` which must be embedded in the message. The call blocks (as a ZIO effect) until the entity calls `replier.reply(value)`. Fails with `SendTimeoutException` if the configured timeout elapses.

```scala
import com.devsisters.shardcake.{ Messenger, Replier }
import scala.util.{ Try, Success, Failure }

// guild: Messenger[GuildMessage]
val joinGuild: ZIO[Any, Throwable, Try[Set[String]]] = 
  guild.send("guild42")(GuildMessage.Join("user99", _))

joinGuild.flatMap {
  case Success(members) => ZIO.debug(s"Joined! Members: $members")
  case Failure(ex)      => ZIO.debug(s"Failed: ${ex.getMessage}") // "Guild is already full!"
}
// Expected output (if guild has < 5 members):
// Joined! Members: Set(user99)
```
```

--------------------------------

### Terminate Local Entity Externally

Source: https://context7.com/devsisters/shardcake/llms.txt

Shows how to externally trigger the termination of a specific entity on the local pod using `Sharding.terminateLocalEntity`. This can be used to evict an entity if necessary.

```scala
Sharding.terminateLocalEntity(Guild, "guild-to-evict")
```

--------------------------------

### Terminate Local Entity from Within Behavior

Source: https://context7.com/devsisters/shardcake/llms.txt

Demonstrates how an entity can self-terminate by calling `Sharding.terminateLocalEntity` from within its behavior. This is useful for completing work and then stopping the entity gracefully. A termination message can be processed before termination.

```scala
import com.devsisters.shardcake.Sharding

// From inside an entity behavior after completing some work:
def behavior(entityId: String, messages: Dequeue[GuildMessage]): RIO[Sharding, Nothing] = 
  messages.take.flatMap {
    case GuildMessage.Terminate(p) =>
      // Signal clean completion then stop this entity
      p.succeed(()) *> Sharding.terminateLocalEntity(Guild, entityId).as(???).orDie
    case msg                       =>
      handleMessage(msg)
  }.forever
```

--------------------------------

### Handle Guild Entity Messages

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Implements the logic for handling `Join` and `Leave` messages for a Guild entity. It uses a `Ref` to manage the set of members and includes logic to prevent joining if the guild is full.

```scala
def handleMessage(state: Ref[Set[String]], message: GuildMessage): RIO[Sharding, Unit] = 
  message match {
    case GuildMessage.Join(userId, replier) =>
      state.get.flatMap(members =>
        if (members.size >= 5)
          replier.reply(Failure(new Exception("Guild is already full!")))
        else
          state.updateAndGet(_ + userId).flatMap {
            newMembers =>
              replier.reply(Success(newMembers))
          }
      )
    case GuildMessage.Leave(userId)         =>
      state.update(_ - userId)
  }
```

--------------------------------

### Register Entity with Termination Message

Source: https://context7.com/devsisters/shardcake/llms.txt

Registers an entity type on the local pod, allowing it to host and process messages. Optionally provides a termination message constructor for graceful entity cleanup before stopping.

```scala
import com.devsisters.shardcake.Sharding
import zio._

// Register Guild entity with a clean termination message
Sharding.registerEntity(
  Guild,
  behavior,
  terminateMessage = (promise: Promise[Nothing, Unit]) => Some(GuildMessage.Terminate(promise)),
  entityMaxIdleTime = Some(5.minutes) // override global idle timeout for this entity type
)
// terminationMessage: entity receives GuildMessage.Terminate, completes the promise, then ZIO.interrupt
// entityMaxIdleTime: entity is stopped after 5 minutes without any message
```

--------------------------------

### Add Kubernetes Health Dependency

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

Add the Kubernetes health check library as a dependency to utilize the Kubernetes API for pod health monitoring. This requires a Pods layer from zio-k8s.

```scala
libraryDependencies += "com.devsisters" %% "shardcake-health-k8s" % "2.7.1"
```

--------------------------------

### Add gRPC Protocol Dependency

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

To use the gRPC protocol for inter-pod communication, add this dependency. This enables the GrpcPods implementation of the Pods trait.

```scala
libraryDependencies += "com.devsisters" %% "shardcake-protocol-grpc" % "2.7.1"
```

--------------------------------

### Sharding.terminateLocalEntity

Source: https://context7.com/devsisters/shardcake/llms.txt

Force-stop a locally-hosted entity by its ID. If a termination message was registered, it is sent first; otherwise, the entity is interrupted immediately. This is typically called from within the entity's own behavior to self-terminate, but can also be called externally to evict an entity from the local pod.

```APIDOC
## Sharding.terminateLocalEntity — Stop a Specific Entity from Within

Force-stop a locally-hosted entity by ID. If a termination message was registered, it is sent first; otherwise the entity is interrupted immediately. Typically called from within the entity's own behavior to self-terminate.

```scala
import com.devsisters.shardcake.Sharding

// From inside an entity behavior after completing some work:
def behavior(entityId: String, messages: Dequeue[GuildMessage]): RIO[Sharding, Nothing] = 
  messages.take.flatMap {
    case GuildMessage.Terminate(p) =>
      // Signal clean completion then stop this entity
      p.succeed(()) *> Sharding.terminateLocalEntity(Guild, entityId).as(???).orDie
    case msg                       => 
      handleMessage(msg)
  }.forever

// Can also be called externally to evict an entity from the local pod:
Sharding.terminateLocalEntity(Guild, "guild-to-evict")
```
```

--------------------------------

### Define Entity Behavior in Scala

Source: https://context7.com/devsisters/shardcake/llms.txt

Implement the entity's behavior function, which processes incoming messages. Use a Ref for state and replier.reply() to send responses. The behavior function runs indefinitely, consuming messages from a queue.

```scala
import com.devsisters.shardcake.{ Sharding, Replier }
import zio.{ Dequeue, Promise, RIO, Ref, ZIO }
import zio.stream.ZStream
import scala.util.{ Failure, Success }

def handleMessage(state: Ref[Set[String]], message: GuildMessage): RIO[Sharding, Unit] =
  message match {
    case GuildMessage.Join(userId, replier) =>
      state.get.flatMap {
        members =>
          if (members.size >= 5)
            replier.reply(Failure(new Exception("Guild is already full!")))
          else
            state.updateAndGet(_ + userId).flatMap(newMembers => replier.reply(Success(newMembers)))
      }
    case GuildMessage.Leave(userId) =>
      state.update(_ - userId)
    case GuildMessage.GetMembers(replier) =>
      state.get.flatMap(members => replier.replyStream(ZStream.fromIterable(members)))
    case GuildMessage.Terminate(p) =>
      p.succeed(()) *> ZIO.interrupt // cleanly stop the entity
  }

// The behavior: loop forever, taking one message at a time
def behavior(entityId: String, messages: Dequeue[GuildMessage]): RIO[Sharding, Nothing] =
  Ref.make(Set.empty[String]).flatMap {
    state =>
      messages.take.flatMap(handleMessage(state, _)).forever
  }
```

--------------------------------

### Define Guild Entity Behavior

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Defines the behavior for the Guild entity. It initializes an empty set of members and continuously processes incoming messages using the `handleMessage` function.

```scala
def behavior(entityId: String, messages: Queue[GuildMessage]): RIO[Sharding, Nothing] = 
  Ref
    .make(Set.empty[String])
    .flatMap(state => messages.take.flatMap(handleMessage(state, _)).forever)
```

--------------------------------

### Registering an Entity Type

Source: https://context7.com/devsisters/shardcake/llms.txt

This function registers an entity type on a specific pod, allowing that pod to host and process messages for the entity. It supports optional termination messages for graceful entity cleanup.

```APIDOC
## `Sharding.registerEntity` — Register an Entity Type on a Pod

Register an entity type so the local pod can host and process messages for that type. Optionally provide a `terminationMessage` constructor so entities can clean up state (e.g., flush to DB) before being stopped during a rebalance or idle timeout.

```scala
import com.devsisters.shardcake.Sharding
import zio._

// Register Guild entity with a clean termination message
Sharding.registerEntity(
  Guild,
  behavior,
  terminateMessage = (promise: Promise[Nothing, Unit]) => Some(GuildMessage.Terminate(promise)),
  entityMaxIdleTime = Some(5.minutes) // override global idle timeout for this entity type
)
// terminationMessage: entity receives GuildMessage.Terminate, completes the promise, then ZIO.interrupt
// entityMaxIdleTime: entity is stopped after 5 minutes without any message
```
```

--------------------------------

### Register Pod Lifecycle with Sharding.registerScoped

Source: https://context7.com/devsisters/shardcake/llms.txt

Notifies the Shard Manager that the pod is ready for shard assignments. Automatically calls unregister on scope closure, triggering a rebalance.

```scala
import com.devsisters.shardcake.Sharding
import zio._

val program: RIO[Sharding with Scope, Unit] = 
  for {
    _ <- Sharding.registerEntity(Guild, behavior)
    _ <- Sharding.registerScoped  // registers on start, unregisters on scope close
    // ... rest of application
  } yield ()
```

--------------------------------

### Send Message and Await Reply with Messenger.send

Source: https://context7.com/devsisters/shardcake/llms.txt

Sends a message expecting a single response, embedding a Replier for the entity to call. Blocks until the entity replies or the configured timeout elapses.

```scala
import com.devsisters.shardcake.{ Messenger, Replier }
import scala.util.{ Try, Success, Failure }

// guild: Messenger[GuildMessage]
val joinGuild: ZIO[Any, Throwable, Try[Set[String]]] = 
  guild.send("guild42")(GuildMessage.Join("user99", _))

joinGuild.flatMap {
  case Success(members) => ZIO.debug(s"Joined! Members: $members")
  case Failure(ex)      => ZIO.debug(s"Failed: ${ex.getMessage}") // "Guild is already full!"
}
// Expected output (if guild has < 5 members):
// Joined! Members: Set(user99)
```

--------------------------------

### Messenger.sendAndReceiveStream

Source: https://context7.com/devsisters/shardcake/llms.txt

Send a single message and receive a ZStream of responses. This is useful for scenarios like subscriptions, paginated results, or live data feeds where multiple replies are expected.

```APIDOC
## `Messenger.sendAndReceiveStream` — Send a Message and Receive a Stream of Replies

Send a single message and get back a `ZStream` of responses. The entity uses a `StreamReplier[A]` to push multiple values. Useful for subscriptions, paginated results, or live data feeds.

```scala
import com.devsisters.shardcake.StreamReplier
import zio.stream.ZStream

// guild: Messenger[GuildMessage]
val listMembers: ZIO[Any, Throwable, Unit] = 
  guild
    .sendAndReceiveStream("guild42")(GuildMessage.GetMembers(_))
    .flatMap(stream => stream.foreach(member => ZIO.debug(s"Member: $member")))

// In the entity behavior, reply with a stream:
// case GuildMessage.GetMembers(replier) =>
//   state.get.flatMap(members => replier.replyStream(ZStream.fromIterable(members)))

// Expected output:
// Member: alice
// Member: bob
// Member: carol
```
```

--------------------------------

### Define Serialization Trait

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

Defines the interface for serializing and deserializing messages between pods. Implement this trait for custom serialization logic.

```scala
trait Serialization {
  def encode(message: Any): Task[Array[Byte]]
  def decode[A](bytes: Array[Byte]): Task[A]
}
```

--------------------------------

### Define Entity Messages and EntityType in Scala

Source: https://context7.com/devsisters/shardcake/llms.txt

Define all possible messages an entity can receive and register the entity type with a unique string identifier. Messages expecting a reply should include a Replier or StreamReplier.

```scala
import com.devsisters.shardcake.{ EntityType, Replier, StreamReplier }
import zio.stream.ZStream
import scala.util.Try

// Define all messages the Guild entity can receive
sealed trait GuildMessage
object GuildMessage {
  case class Join(userId: String, replier: Replier[Try[Set[String]]]) extends GuildMessage
  case class Leave(userId: String)                                    extends GuildMessage
  case class GetMembers(replier: StreamReplier[String])               extends GuildMessage
  case class Terminate(p: zio.Promise[Nothing, Unit])                 extends GuildMessage
}

// Register entity type with a unique string identifier
object Guild extends EntityType[GuildMessage]("guild")
```

--------------------------------

### Define Guild Entity Messages

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Defines the sealed trait `GuildMessage` and its case classes `Join` and `Leave` for entity communication. The `Join` message includes a `Replier` for sending responses.

```scala
sealed trait GuildMessage

object GuildMessage {
  case class Join(userId: String, replier: Replier[Try[Set[String]]]) extends GuildMessage
  case class Leave(userId: String)                                    extends GuildMessage
}
```

--------------------------------

### Define PodsHealth Trait

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

Defines the interface for checking the health status of pods. Implement this trait to integrate custom health checking mechanisms.

```scala
trait PodsHealth {
  def isAlive(podAddress: PodAddress): UIO[Boolean]
}
```

--------------------------------

### Define Guild Entity Type

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/README.md

Defines the `Guild` entity type with a unique identifier 'guild'. This establishes the entity's type within the Shardcake system.

```scala
object Guild extends EntityType[GuildMessage]("guild")
```

--------------------------------

### Define Pods Trait

Source: https://github.com/devsisters/shardcake/blob/series/2.x/vuepress/docs/docs/customization.md

The Pods trait defines the communication interface for remote pods. It includes methods for assigning/unassigning shards, pinging pods, and sending messages.

```scala
trait Pods {
  def assignShards(pod: PodAddress, shards: Set[ShardId]): Task[Unit]
  def unassignShards(pod: PodAddress, shards: Set[ShardId]): Task[Unit]
  def ping(pod: PodAddress): Task[Unit]
  
  def sendMessage(pod: PodAddress, message: BinaryMessage): Task[Option[Array[Byte]]]
  def sendMessageStreaming(pod: PodAddress, message: BinaryMessage): ZStream[Any, Throwable, Array[Byte]]
}
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.