diff --git a/.gitignore b/.gitignore index 21df78e4e..177d3db1c 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,5 @@ node_modules .vitepress/dist .idea .vscode -.DS_Store \ No newline at end of file +.DS_Store +.opencode \ No newline at end of file diff --git a/.vitepress/config/en.ts b/.vitepress/config/en.ts index 7379091f5..19c8b258a 100644 --- a/.vitepress/config/en.ts +++ b/.vitepress/config/en.ts @@ -121,7 +121,7 @@ function nav(): DefaultTheme.NavItem[] { { text: 'v1.16', link: 'https://v116.goravel.dev/' - }, + } ] }, { @@ -179,6 +179,10 @@ function sidebarPrologue(): DefaultTheme.SidebarItem[] { function sidebarUpgrade(): DefaultTheme.SidebarItem[] { return [ + { + text: 'Upgrading To v1.19 From v1.18', + link: 'v1.19' + }, { text: 'Upgrading To v1.18 From v1.17', link: 'v1.18' @@ -274,6 +278,10 @@ function sidebarAdvanced(): DefaultTheme.SidebarItem[] { text: 'Event', link: 'event' }, + { + text: 'Broadcasting', + link: 'broadcasting' + }, { text: 'File Storage', link: 'filesystem' diff --git a/en/digging-deeper/broadcasting.md b/en/digging-deeper/broadcasting.md new file mode 100644 index 000000000..ebf87e46b --- /dev/null +++ b/en/digging-deeper/broadcasting.md @@ -0,0 +1,401 @@ +# Broadcasting + +[[toc]] + +## Introduction + +Goravel's broadcasting allows you to push realtime, live-updating data to your frontend over WebSockets. Instead of the client polling the server for changes, your backend broadcasts events to named channels, and subscribed clients receive them instantly. + +The core concepts are simple: clients connect to channels on the frontend, while your Goravel application broadcasts events to these channels on the backend. + +## Installation + +The broadcasting facade is not installed by default. Install it with the `package:install` command: + +```shell +./artisan package:install Broadcast +``` + +This creates `config/broadcasting.go` and registers the `broadcasting.ServiceProvider` in `bootstrap/providers.go`. + +### Supported Drivers + +| Driver | Description | +| -------- | ------------------------------------------------------------------------------------------------------------------------------ | +| `pusher` | Broadcasts to any [Pusher](https://pusher.com/channels) protocol compatible server, such as [Soketi](https://docs.soketi.app/) | +| `log` | Writes broadcasts to the log, useful for local development | +| `null` | Discards all broadcasts, useful for testing | + +## Defining Broadcast Events + +Use the `make:event` Artisan command with the `--broadcast` flag to scaffold an event that implements the `ShouldBroadcast` contract: + +```shell +./artisan make:event OrderShipped --broadcast +./artisan make:event OrderShipped --broadcast --now +``` + +- `--broadcast` scaffolds an event with `BroadcastOn`, `BroadcastAs`, `BroadcastWith`, and `BroadcastWhen` methods, including a `var _ broadcasting.ShouldBroadcast = (*OrderShipped)(nil)` compile-time assertion. +- Adding `--now` also scaffolds a `BroadcastNow() bool` method returning `true`, so the event broadcasts synchronously instead of through the queue. + +Broadcast Events implement the `ShouldBroadcast` contract from `contracts/broadcasting`. It requires four methods: + +- `BroadcastOn() []string` — the channel names to broadcast on. +- `BroadcastAs() string` — the event name. +- `BroadcastWith() map[string]any` — the event payload. +- `BroadcastWhen() bool` — whether the event should be broadcast. + +```go +package events + +import ( + "strconv" + + "github.com/goravel/framework/broadcasting" +) + +type OrderShipped struct { + OrderID int +} + +func (e *OrderShipped) BroadcastOn() []string { + return []string{ + broadcasting.PrivateChannel("orders." + strconv.Itoa(e.OrderID)), + } +} + +func (e *OrderShipped) BroadcastAs() string { + return "order.shipped" +} + +func (e *OrderShipped) BroadcastWith() map[string]any { + return map[string]any{"order": map[string]any{"id": e.OrderID}} +} + +func (e *OrderShipped) BroadcastWhen() bool { + return true +} +``` + +`BroadcastOn` returns the channel names to broadcast to, built with the `PublicChannel`, `PrivateChannel`, and `PresenceChannel` helpers. Each helper returns a plain `string` channel name: + +| Helper | Returns (`string`) | Description | +| ----------------- | ------------------ | -------------------------------------------- | +| `PublicChannel` | `orders` | Anyone can subscribe | +| `PrivateChannel` | `private-orders` | Requires authentication and authorization | +| `PresenceChannel` | `presence-orders` | Like private, plus exposes who is subscribed | + +### Optional Contracts + +Implement these optional contracts to customize how events are broadcast: + +| Contract | Method | Purpose | +| ------------------------------------ | ------------------------------------ | -------------------------------------------------------------------------------------------------------------------- | +| `ShouldBroadcastNow` | `BroadcastNow() bool` | Broadcast synchronously instead of via the queue | +| `ShouldBroadcastWithQueue` | `BroadcastQueue() string` | The queue name to use | +| `ShouldBroadcastWithQueueConnection` | `BroadcastQueueConnection() string` | The queue connection to use | +| `ShouldBroadcastWithConnections` | `BroadcastConnections() []string` | The broadcast connections to use | +| `ShouldBroadcastWithDelay` | `BroadcastDelay() time.Time` | Delay the broadcast until a given time | +| `ShouldBroadcastWithTimeout` | `BroadcastTimeout() time.Duration` | Bound how long a broadcast may take | +| `ShouldBroadcastWithTries` | `BroadcastTries() int` | The maximum number of attempts for the queued broadcast; `0` (or not implementing the contract) means single-shot | +| `ShouldBroadcastWithBackoff` | `BroadcastBackoff() []time.Duration` | The delay before each retry attempt, in order; the last value repeats. Only effective together with `BroadcastTries` | + +```go +func (e *OrderShipped) BroadcastNow() bool { + return true +} + +func (e *OrderShipped) BroadcastQueue() string { + return "broadcasts" +} + +// Retry the queued broadcast up to 3 attempts, waiting 10s then 30s +// between attempts (the last backoff value repeats for later attempts). +func (e *OrderShipped) BroadcastTries() int { + return 3 +} + +func (e *OrderShipped) BroadcastBackoff() []time.Duration { + return []time.Duration{10 * time.Second, 30 * time.Second} +} +``` + +By default, broadcasts are dispatched as [queued jobs](queues.md), so make sure a queue worker is running. Implementing `ShouldBroadcastNow` skips the queue. + +## Authorizing Channels + +Private and presence channels require authorization. Goravel automatically registers the `/broadcasting/auth` route (configured in the `auth` section of `config/broadcasting.go`) to handle authorization requests. + +Register authorization callbacks in `routes/channels.go` with `facades.Broadcast().Channel()`. The callback receives the authenticated user ID and any `{param}` wildcards from the channel name, and returns `(authorized, userInfo)`: + +```go +package routes + +import ( + "context" + + "goravel/app/facades" +) + +func Channels() { + facades.Broadcast().Channel("orders.{orderId}", func(ctx context.Context, userID any, channelName string, params map[string]string) (bool, any) { + return userID != nil && params["orderId"] != "", nil + }) +} +``` + +The second return value is the user info broadcast to other subscribers, and is used by presence channels. + +Call `Channels()` during the application bootstrap: + +```go +func Boot() contractsfoundation.Application { + return foundation.Setup(). + WithRouting(func() { + // ... + routes.Channels() + }). + // ... +} +``` + +### Channel Classes + +For many channels, extract authorization into a class using the `make:channel` command: + +```shell +./artisan make:channel OrderChannel +``` + +This generates a `ChannelAuthFunc` in `app/broadcasting`: + +```go +package broadcasting + +import ( + "context" + + "github.com/goravel/framework/contracts/broadcasting" +) + +func OrderChannel(ctx context.Context, userID any, channelName string, params map[string]string) (bool, any) { + return false, nil +} + +var _ broadcasting.ChannelAuthFunc = OrderChannel +``` + +Register it in `routes/channels.go`: + +```go +facades.Broadcast().Channel("orders.{orderId}", appbroadcasting.OrderChannel) +``` + +## Dispatching Events + +Dispatch an event with `facades.Broadcast().Dispatch()`. The event is broadcast if it implements `ShouldBroadcast`: + +```go +package controllers + +import ( + "context" + + "github.com/goravel/framework/contracts/http" + + "goravel/app/events" + "goravel/app/facades" +) + +func (c *OrderController) Ship(ctx http.Context) http.Response { + err := facades.Broadcast().Dispatch(context.Background(), &events.OrderShipped{ + OrderID: 1, + }) + if err != nil { + return ctx.Response().String(http.StatusInternalServerError, err.Error()) + } + + return ctx.Response().Success().Json(http.Json{"message": "shipped"}) +} +``` + +## Receiving Broadcasts + +Since Goravel's broadcast drivers speak the [Pusher protocol](https://pusher.com/docs/channels/library_auth_reference/pusher-websockets-protocol/), you can receive broadcasts with [Laravel Echo](#laravel-echo) or with a raw WebSocket client. + +### Laravel Echo + +Install [Laravel Echo](https://github.com/laravel/echo), which wraps the Pusher protocol and works with any Pusher compatible server: + +```shell +npm install laravel-echo pusher-js +``` + +```js +import Echo from 'laravel-echo' +import Pusher from 'pusher-js' + +window.Pusher = Pusher + +window.Echo = new Echo({ + broadcaster: 'pusher', + key: import.meta.env.VITE_PUSHER_APP_KEY, + wsHost: import.meta.env.VITE_PUSHER_HOST, + wsPort: import.meta.env.VITE_PUSHER_PORT, + forceTLS: false +}) +``` + +Subscribe to a channel and listen for events: + +```js +Echo.channel(`orders.${orderId}`).listen('.order.shipped', (e) => { + console.log(e.order) +}) +``` + +Note the leading `.` in `.order.shipped`: it tells Echo to use the name from `BroadcastAs` as-is instead of prepending a namespace. + +To leave a channel, use `leave`: + +```js +Echo.leave(`orders.${orderId}`) +``` + +### Raw WebSocket + +If you don't want to use Echo, you can connect directly with any Pusher protocol WebSocket client. The following examples use the browser's native `WebSocket` API. + +#### Connecting + +Connect to `/app/{key}`. The connection URL uses your Pusher `key` and the WebSocket host from your broadcasting configuration: + +```js +const ws = new WebSocket( + 'ws://127.0.0.1:6001/app/test-key?protocol=7&client=js&version=7.0.0' +) + +let socketId = '' + +ws.onmessage = (evt) => { + const msg = JSON.parse(evt.data) + + // Save the socket ID from the connection handshake, it is required for + // authorizing private and presence channels. + if (msg.event === 'pusher:connection_established') { + socketId = JSON.parse(msg.data).socket_id + } + + // Handle your application's broadcast events. + if (msg.event === 'order.shipped') { + console.log(JSON.parse(msg.data)) + } +} +``` + +#### Subscribing to Public Channels + +Send a `pusher:subscribe` message with the channel name. No authorization is needed: + +```js +ws.send( + JSON.stringify({ + event: 'pusher:subscribe', + data: { channel: 'orders.1' } + }) +) +``` + +#### Subscribing to Private Channels + +Private and presence channels require authorization. First obtain a token from the `/broadcasting/auth` endpoint, posting your `socket_id` and `channel_name`. The request must be authenticated as the current user: + +```js +const resp = await fetch('/broadcasting/auth', { + method: 'POST', + headers: { + 'Content-Type': 'application/x-www-form-urlencoded', + Authorization: jwtToken + }, + body: `socket_id=${encodeURIComponent(socketId)}&channel_name=${encodeURIComponent(channelName)}` +}) + +const { auth, channel_data } = await resp.json() +``` + +Then subscribe with the returned token (and `channel_data` for presence channels): + +```js +ws.send( + JSON.stringify({ + event: 'pusher:subscribe', + data: { + channel: channelName, + auth, + ...(channel_data ? { channel_data } : {}) + } + }) +) +``` + +## Presence Channels + +Presence channels are private channels that also expose who is subscribed, which makes collaborative features like "who is viewing this page" easy to build. + +Authorize presence channels and return the user info to broadcast to other subscribers: + +```go +func Channels() { + facades.Broadcast().Channel("team.{teamId}", func(ctx context.Context, userID any, channelName string, params map[string]string) (bool, any) { + if userID == nil || params["teamId"] == "" { + return false, nil + } + + return true, map[string]any{"id": userID, "name": "Alice"} + }) +} +``` + +Broadcast to a presence channel by returning a `PresenceChannel` from `BroadcastOn`: + +```go +func (e *TeamCreated) BroadcastOn() []string { + return []string{ + broadcasting.PresenceChannel("team." + strconv.Itoa(e.TeamID)), + } +} +``` + +Join a presence channel with Echo's `join` method: + +```js +Echo.join(`team.${teamId}`) + .here((users) => { + console.log(users) + }) + .joining((user) => { + console.log(user.name) + }) + .leaving((user) => { + console.log(user.name) + }) +``` + +With a raw WebSocket client, members are reported through the `pusher_internal:member_added` and `pusher_internal:member_removed` events, and the initial member list arrives in the `pusher_internal:subscription_succeeded` message: + +```js +ws.onmessage = (evt) => { + const msg = JSON.parse(evt.data) + const data = typeof msg.data === 'string' ? JSON.parse(msg.data) : msg.data + + if (msg.event === 'pusher_internal:subscription_succeeded') { + console.log('Members:', data.presence.ids) + } + if (msg.event === 'pusher_internal:member_added') { + console.log('Joined:', data.user_info) + } + if (msg.event === 'pusher_internal:member_removed') { + console.log('Left:', data.user_id) + } +} +``` diff --git a/en/digging-deeper/queues.md b/en/digging-deeper/queues.md index 44ca4b28b..82b59b77f 100644 --- a/en/digging-deeper/queues.md +++ b/en/digging-deeper/queues.md @@ -51,6 +51,8 @@ After implementing the custom driver, you can add the configuration to `config/q "driver": "custom", "connection": "default", "queue": "default", + "concurrent": 5, + "retry_after": 60, "via": func() (queue.Driver, error) { return redisfacades.Queue("redis") // The redis value is the key of connections }, @@ -91,6 +93,23 @@ func (d *KafkaDriver) Receive(ctx context.Context, queue string, count int) ([]q When `Receive` is available, the worker runs a blocking batch loop with a 5-second per-call timeout and exponential backoff (100ms–3.2s) on errors or empty batches. The `context.Context` is canceled on worker shutdown, ensuring clean termination. +### retry_after + +The `retry_after` option (default `60`, in seconds) is available on every connection and controls the crashed-worker reservation-expiry window. If a worker crashes while holding a job, its reservation expires after `retry_after` seconds and the job is recovered by other workers. It must exceed the maximum job runtime to avoid double-processing long-running jobs: + +```go +"database": map[string]any{ + "driver": "database", + "connection": "sqlite", + "queue": "default", + "concurrent": 5, + // Reservation expiry for crashed workers; must exceed the maximum job runtime + "retry_after": 60, +}, +``` + +Custom drivers (including the Redis driver) read this option from each connection's config. + ## Creating Jobs ### Generating Job Classes @@ -138,7 +157,10 @@ func (r *ProcessPodcast) Handle(args ...any) error { #### Job Retry -Job classes support an optional `ShouldRetry(err error, attempt int) (retryable bool, delay time.Duration)` method, which is used to control job retry. +Job classes support an optional `ShouldRetry(err error, attempt int) (retryable bool, delay time.Duration)` method, which is used to control job retry. When a job implementing it fails, the queue worker **releases** the job back to the queue instead of immediately retrying it in-memory. The attempt count is persisted with the reservation, so retries survive worker restarts, respect the release delay, and can be picked up by any worker. + +- `retryable = true` — the job is released back to the queue and runs again after `delay`, with its attempt count preserved. +- `retryable = false` — the job is marked as failed and recorded in the `failed_jobs` table. ```go // ShouldRetry determines if the job should be retried based on the error. @@ -147,9 +169,23 @@ func (r *ProcessPodcast) ShouldRetry(err error, attempt int) (retryable bool, de } ``` +For example, the following job fails on its first two attempts and succeeds on the third: + +```go +// ShouldRetry retries while the attempt count is within the failure window, +// then gives up and lets the job land in failed_jobs. +func (r *TestRetryable) ShouldRetry(err error, attempt int) (bool, time.Duration) { + if attempt <= 2 { + return true, 100 * time.Millisecond + } + + return false, 0 +} +``` + ## Start Queue Server -The default queue worker will be run by the runner of queue seriver provider, if you want to start multiple queue workers with different configuration, you can create [a runner](../architecture-concepts/service-providers.md#runners) and add it to the `WithRunners` function in the `bootstrap/app.go` file: +The default queue worker will be run by the runner of queue server provider, if you want to start multiple queue workers with different configuration, you can create [a runner](../architecture-concepts/service-providers.md#runners) and add it to the `WithRunners` function in the `bootstrap/app.go` file: ```go func Boot() contractsfoundation.Application { diff --git a/en/getting-started/installation.md b/en/getting-started/installation.md index d97aee581..98b4a73b3 100644 --- a/en/getting-started/installation.md +++ b/en/getting-started/installation.md @@ -4,7 +4,7 @@ ## Server Requirements -- Golang >= 1.23 +- Golang >= 1.26 ## Installation diff --git a/en/prologue/compare-with-laravel.md b/en/prologue/compare-with-laravel.md index d50752d85..bd5123a53 100644 --- a/en/prologue/compare-with-laravel.md +++ b/en/prologue/compare-with-laravel.md @@ -34,7 +34,7 @@ Goravel is heavily inspired by the Laravel framework, aiming to bring similar el | [Testing](https://www.goravel.dev/testing/getting-started.html) | ✅ | ✅ | | | [Validation](https://www.goravel.dev/the-basics/validation.html) | ✅ | ✅ | ctx.Request().ValidateRequest()
$request->validate() | | [View](https://www.goravel.dev/the-basics/views.html) | ✅ | ✅ | ctx.Response().View().Make("welcome.tmpl")
view('welcome') | +| Notifications | ✅ | ✅ | | +| [Broadcasting](https://www.goravel.dev/digging-deeper/broadcasting.html) | ✅ | ✅ | facades.Broadcast().Channel("channel", ...)
Broadcast::channel('channel', ...) | +| Inertia | ✅ | ✅ | | | [Grpc](https://www.goravel.dev/the-basics/grpc.html) | ✅ | 🚧 | | -| Notifications | 🚧 | ✅ | | -| Broadcasting | 🚧 | ✅ | | -| Livewire | 🚧 | ✅ | | diff --git a/en/upgrade/v1.19.md b/en/upgrade/v1.19.md new file mode 100644 index 000000000..0e61a0f02 --- /dev/null +++ b/en/upgrade/v1.19.md @@ -0,0 +1,235 @@ +# Upgrading To v1.19 From v1.18 + +## Exciting New Features 🎉 + +- [Broadcasting](#broadcasting) + +## Enhancements 🚀 + +- [Release-based job retry with crash recovery](#release-based-job-retry-with-crash-recovery) + +## Breaking Changes 🛠 + +- [ReservedJob requires Attempts and Release](#reservedjob-requires-attempts-and-release) + +## Upgrade Guide + +As [Golang v1.25 is no longer maintained](https://endoflife.date/go), the Golang version Goravel supports has been upgraded from 1.25 to 1.26. + +goravel/example project from v1.18 to v1.19 PR can be used as an upgrade reference: [goravel/example#XXX](https://github.com/goravel/example/pull/TODO). + +You can copy and paste the following into your AI coding agent to upgrade your Goravel project automatically. + +````markdown +# Upgrade Guide for Goravel v1.19 + +**Before you begin**: ensure all changes are committed or stashed, the project builds cleanly (`go build ./...`), and Go ≥ 1.26 is installed, Goravel version ≥ 1.18.0. + +## Step 1: Update Go module dependencies (Go ≥ 1.26 required) + +**Detection**: Check which facade packages are already installed: + +```shell +rg -l 'goravel/(gin|fiber|redis|s3|oss|cos|minio)' go.sum 2>/dev/null +``` + +**Action**: Run the framework upgrade command and upgrade only the facade packages detected in `go.sum`: + +```shell +go get github.com/goravel/framework@latest + +# For each facade found in detection, run the corresponding go get. +# If goravel/redis is in go.sum: go get github.com/goravel/redis@latest +# If goravel/gin is in go.sum: go get github.com/goravel/gin@latest +# If goravel/fiber is in go.sum: go get github.com/goravel/fiber@latest +# If goravel/s3 is in go.sum: go get github.com/goravel/s3@latest +# If goravel/oss is in go.sum: go get github.com/goravel/oss@latest +# If goravel/cos is in go.sum: go get github.com/goravel/cos@latest +# If goravel/minio is in go.sum: go get github.com/goravel/minio@latest + +go mod tidy +``` + +## Step 2: Add retry_after to each queue connection + +`retry_after` (default `60`, in seconds) is the crashed-worker reservation-expiry window: if a worker crashes while holding a job, its reservation expires after `retry_after` seconds and the job is recovered by other workers. Add it to each `database` and custom connection in `config/queue.go`: + +```diff + "database": map[string]any{ + "driver": "database", + "connection": "sqlite", + "queue": "default", + "concurrent": 5, ++ // Reservation expiry for crashed workers; must exceed the maximum job runtime ++ "retry_after": 60, + }, +``` + +The value must exceed the maximum job runtime to avoid double-processing long-running jobs. + +### Verification + +```shell +rg -n 'retry_after' config/queue.go + +go build ./... +``` + +## Step 3: Update custom queue drivers to the ReservedJob contract (only if you have custom drivers) + +**Detection**: Check if any queue connection in `config/queue.go` uses a custom driver (a non-built-in `driver` value or a `"via"` key): + +```shell +rg -n '"driver"\s*:\s*"custom"|"via"' config/queue.go +``` + +If no custom drivers are found, skip this step. + +`contracts/queue.ReservedJob` now requires two new methods in addition to `Delete() error` and `Task() Task`: + +- `Attempts() int` — the number of times the job has been attempted so far. It must be persisted with the reservation, so retry decisions survive worker restarts. +- `Release(delay time.Duration) error` — make the job available again after the given delay so it can be retried, incrementing attempts on the next pop. + +```go +func (r *ReservedJob) Attempts() int { + return r.jobRecord.Attempts +} + +// Release removes the job from the reserved set and makes it available +// again after the delay, preserving the serialized attempt count. +func (r *ReservedJob) Release(delay time.Duration) error { + // remove from reserved set and push to the delayed/ready set with score = now + delay + return nil +} +``` + +Also read the connection's `retry_after` config and use it to recover expired reservations left by crashed workers (see the [database driver](https://github.com/goravel/framework/blob/master/queue/driver_database.go) and the [Redis driver](https://github.com/goravel/redis/blob/master/queue.go) implementations). + +### Verification + +```shell +go build ./... +``` + +## Step 4: Bump the jobs table migration to millisecond-precision timestamps (optional) + +For `database` queue connections, update the [20210101000002_create_jobs_table.go](https://github.com/goravel/goravel/blob/master/database/migrations/20210101000002_create_jobs_table.go) migration in `database/migrations` so new databases store `reserved_at`, `available_at`, and `created_at` with millisecond precision, so sub-second delays and retries survive round trips: + +```diff +- table.DateTimeTz("reserved_at").Nullable() +- table.DateTimeTz("available_at") +- table.DateTimeTz("created_at").UseCurrent() ++ table.DateTimeTz("reserved_at", 3).Nullable() ++ table.DateTimeTz("available_at", 3) ++ table.DateTimeTz("created_at", 3).UseCurrent() +``` + +Existing databases do not need this migration unless you rely on sub-second retry delays. + +## Step 5: Final verification + +After completing all steps, run the full verification suite: + +```shell +go build ./... +go vet ./... +go test ./... 2>&1 || true +``` + +If tests fail, the most common causes are: + +- Custom queue drivers missing `Attempts()` or `Release()` on `ReservedJob` (step 3). +```` + +## Feature Introduction + +### Broadcasting + +Goravel v1.19 introduces a first-party broadcasting module for pushing realtime, live-updating data to your frontend over WebSockets. Instead of the client polling the server for changes, your backend broadcasts events to named channels, and subscribed clients receive them instantly. + +Install the facade with the `package:install` command: + +```shell +./artisan package:install Broadcast +``` + +The initial release includes: + +- Pusher-protocol broadcasting with `pusher`, `log`, and `null` drivers — compatible with any Pusher protocol server such as [Soketi](https://docs.soketi.app/). +- Events implementing the `ShouldBroadcast` contract dispatched via `facades.Broadcast().Dispatch()`, with `BroadcastOn` channels, `BroadcastAs` event names, `BroadcastWith` payloads, and `BroadcastWhen` conditionals. +- Public, private, and presence channels, authorization callbacks via `facades.Broadcast().Channel()`, and the `make:channel` command to extract authorization into channel classes. +- Receiving broadcasts on the frontend with [Laravel Echo](https://github.com/laravel/echo) or any raw Pusher protocol WebSocket client. + +```go +err := facades.Broadcast().Dispatch(context.Background(), &events.OrderShipped{ + OrderID: 1, +}) +``` + +[View Document](../digging-deeper/broadcasting.md#dispatching-events) + +### Release-based job retry with crash recovery + +Jobs implementing `ShouldRetry(err error, attempt int) (retryable bool, delay time.Duration)` are now released back to the queue on failure instead of being retried in-memory ([goravel/framework#1531](https://github.com/goravel/framework/pull/1531), [goravel/redis#149](https://github.com/goravel/redis/pull/149)). The attempt count is persisted with the reservation, so: + +- Retries survive worker restarts and can be picked up by any worker. +- The release delay is respected across workers, with sub-second precision for the database and Redis drivers. +- `retryable = false` lands the job in the `failed_jobs` table. + +```go +// Retry while the attempt count is within the failure window, then give up. +func (r *TestRetryable) ShouldRetry(err error, attempt int) (bool, time.Duration) { + if attempt <= 2 { + return true, 100 * time.Millisecond + } + + return false, 0 +} +``` + +Queue connections also support a `retry_after` configuration option (default `60`, in seconds) that controls the crashed-worker reservation-expiry window. If a worker crashes while holding a job, its reservation expires after `retry_after` seconds and the job is recovered by other workers. The value must exceed the maximum job runtime to avoid double-processing long-running jobs: + +```go +"database": map[string]any{ + "driver": "database", + "connection": "sqlite", + "queue": "default", + "concurrent": 5, + // Reservation expiry for crashed workers; must exceed the maximum job runtime + "retry_after": 60, +}, +``` + +The `jobs` table migration now stores `reserved_at`, `available_at`, and `created_at` with millisecond precision, so sub-second delays and release-based retries survive round trips to the database: + +```go +table.DateTimeTz("reserved_at", 3).Nullable() +table.DateTimeTz("available_at", 3) +table.DateTimeTz("created_at", 3).UseCurrent() +``` + +New installations get this automatically. Existing databases can keep their second-precision columns unless you rely on sub-second retry delays. + +[View Document](../digging-deeper/queues.md#job-retry) + +### ReservedJob requires Attempts and Release + +`contracts/queue.ReservedJob` now requires two new methods in addition to `Delete() error` and `Task() Task`: + +```go +type ReservedJob interface { + // Attempts returns the number of times the job has been attempted so far. + Attempts() int + // Delete removes the job from the queue. + Delete() error + // Release makes the job available again after the given delay so it can + // be retried, incrementing attempts on the next pop. + Release(delay time.Duration) error + // Task returns the task to execute. + Task() Task +} +``` + +Custom queue drivers must be updated to implement `Attempts()` and `Release()`. See the [Upgrade Guide](#upgrade-guide) above for the full migration steps. + +[View Document](../digging-deeper/queues.md)