---
title: "Authoring adapters"
description: "Build custom watchers, stores, queue integrations, and live-update fanout."
---

> Documentation Index
> Fetch the complete documentation index at: https://adonisjs-periscope.pages.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Authoring adapters

Periscope exposes narrow extension seams for host integrations, storage, and live dashboard fanout.
The public contracts are exported from `@rikology/adonisjs-periscope`.

## Common lifecycle rules

- Adapter and watcher names are stable identifiers. Use the same value in emitted event metadata.
- `register` may be synchronous or asynchronous and may return a cleanup function. Cleanup can also
  be asynchronous and must be safe to call more than once.
- Registration, event handling, and cleanup are best effort. A Periscope integration must never
  throw into the queue worker, scheduler, notification sender, socket server, or other host path.
  Catch observer calls at callbacks controlled by the host library and degrade to a dropped signal.
- Avoid importing optional peers at module evaluation time. Resolve them during `register` and
  return without subscribing when absent.
- Roll back partial registration when setup fails. Bound metadata maps and background work, and
  settle pending work during cleanup.
- Payload capture is opt-in. Respect `options.capturePayload`; Periscope still applies bounded
  serialization and redaction after an adapter emits a value.

The shipped BullMQ adapter demonstrates these rules: job lookups have a timeout, tracking maps have
a ceiling, event callbacks catch observer failures, registration closes partially opened resources,
and cleanup uses `Promise.allSettled`.

## Queue adapters

`QueueWatcherAdapter` bridges a queue library to the `job_schedule` watcher:

```ts
interface QueueWatcherAdapter {
  readonly name: string
  register(
observer: QueueWatcherObserver,
options?: { capturePayload: boolean }
  ): void | (() => void | Promise<void>) | Promise<void | (() => void | Promise<void>)>
}
```

The observer accepts `started`, `completed`, `failed`, and `scheduled` lifecycle events. Every
`QueueJobEvent` requires `adapter`, `queue`, and `jobId`; it may include `name`, `payload`,
`attempts`, `scheduledAt`, `durationMs`, and `correlationId`. Completed and failed events may add
`result` or `error`.

Use the optional correlation methods when the queue library exposes the necessary interception:

- Call `observer.dispatching(event)` before serialization, persist its returned `correlationId` in
  transport metadata, and echo it from worker lifecycle events. The producer's dispatch entry and
  worker entries can then share a batch ID across processes.
- Call `observer.wrapJob(event, run)` around the actual handler execution when middleware can wrap
  it. Queries, logs, and outgoing requests made by `run` then inherit the queue batch. Do not retry
  `run` if it throws; that error belongs to the job.

A compact event-emitter adapter looks like this:

```ts
import type {
  QueueJobEvent,
  QueueWatcherAdapter,
  QueueWatcherObserver,
} from '@rikology/adonisjs-periscope'

export class AcmeQueueAdapter implements QueueWatcherAdapter {
  readonly name = 'acme-queue'

  register(observer: QueueWatcherObserver, options = { capturePayload: false }) {
const notify = (callback: () => void) => {
  try {
    callback()
  } catch {
    // Never reject into Acme Queue.
  }
}
const toEvent = (job: AcmeJob): QueueJobEvent => ({
  adapter: this.name,
  queue: job.queue,
  jobId: job.id,
  name: job.name,
  attempts: job.attempts,
  ...(options.capturePayload ? { payload: job.payload } : {}),
  ...(job.meta.periscopeId ? { correlationId: job.meta.periscopeId } : {}),
})

const onDispatch = (job: AcmeJob) => {
  try {
    const correlation = observer.dispatching?.(toEvent(job))
    if (correlation) job.meta.periscopeId = correlation.correlationId
  } catch {
    // Dispatch proceeds without correlation.
  }
}
const onStarted = (job: AcmeJob) => notify(() => observer.started(toEvent(job)))
const onCompleted = (job: AcmeJob, result: unknown) =>
  notify(() =>
    observer.completed({
      ...toEvent(job),
      ...(options.capturePayload ? { result } : {}),
    })
  )
const onFailed = (job: AcmeJob, error: unknown) =>
  notify(() => observer.failed({ ...toEvent(job), error }))

acmeQueue.on('dispatching', onDispatch)
acmeQueue.on('started', onStarted)
acmeQueue.on('completed', onCompleted)
acmeQueue.on('failed', onFailed)

let active = true
return () => {
  if (!active) return
  active = false
  acmeQueue.off('dispatching', onDispatch)
  acmeQueue.off('started', onStarted)
  acmeQueue.off('completed', onCompleted)
  acmeQueue.off('failed', onFailed)
}
  }
}
```

Register queue adapters in `watchers.job_schedule.adapters`:

```ts
import { BullQueueAdapter } from '@rikology/adonisjs-periscope/watchers/bull_queue'

export default defineConfig({
  watchers: {
job_schedule: {
  enabled: true,
  adapters: [
    new BullQueueAdapter({
      queues: [{ name: 'mail', connection: { host: '127.0.0.1', port: 6379 } }],
    }),
  ],
  capturePayload: false,
},
  },
})
```

`BullQueueAdapter` observes BullMQ through `QueueEvents` and enriches lifecycle entries with job
name, attempts, and opt-in payload through bounded, best-effort job lookups. `AdonisQueueAdapter`
observes the experimental `@adonisjs/queue` tracing channels. Both are silent no-ops when their
queue package is absent, observe existing workers, and never replace application workers.

## Scheduler adapters

`SchedulerWatcherAdapter` represents actual cron or task executions, not delayed queue jobs:

```ts
interface SchedulerWatcherAdapter {
  readonly name: string
  register(
observer: SchedulerWatcherObserver,
options?: { capturePayload: boolean }
  ): void | (() => void | Promise<void>) | Promise<void | (() => void | Promise<void>)>
}
```

Call `taskStarted`, followed by `taskCompleted` or `taskFailed`. Each event requires `adapter` and a
stable `task`; `schedule`, `runId`, and `durationMs` are optional. A result can include `result` or
`error`. `wrapTask(event, run)` places work performed by the handler in the task context. Scheduler
lifecycles are stored as `schedule` entries and grouped in `schedule` batches.

```ts
export class AcmeSchedulerAdapter implements SchedulerWatcherAdapter {
  readonly name = 'acme-scheduler'

  register(observer: SchedulerWatcherObserver, options = { capturePayload: false }) {
const execute = async (task: AcmeTask) => {
  const event = {
    adapter: this.name,
    task: task.name,
    schedule: task.cron,
    runId: task.runId,
  }

  try {
    observer.taskStarted(event)
  } catch {}

  const startedAt = performance.now()
  try {
    const result = observer.wrapTask
      ? await observer.wrapTask(event, () => task.run())
      : await task.run()
    try {
      observer.taskCompleted({
        ...event,
        durationMs: performance.now() - startedAt,
        ...(options.capturePayload ? { result } : {}),
      })
    } catch {}
    return result
  } catch (error) {
    try {
      observer.taskFailed({ ...event, durationMs: performance.now() - startedAt, error })
    } catch {}
    throw error
  }
}

return acmeScheduler.use(execute)
  }
}
```

Register scheduler adapters separately from queues:

```ts
watchers: {
  job_schedule: {
enabled: true,
schedulers: [new AcmeSchedulerAdapter()],
capturePayload: false,
  },
},
```

## Notification adapters

`NotificationWatcherAdapter` reports delivery outcomes:

```ts
interface NotificationWatcherAdapter {
  readonly name: string
  register(
observer: NotificationWatcherObserver,
options?: { capturePayload: boolean }
  ): void | (() => void | Promise<void>) | Promise<void | (() => void | Promise<void>)>
}
```

Call `sent` or `failed` with `adapter`, `channel`, and `notification`. Optional fields are
`notifiable`, `payload`, `durationMs`, and, for failures, `error`. `notifiable` must be a scalar ID
or already-masked address, never a user model.

```ts
export class AcmeNotificationAdapter implements NotificationWatcherAdapter {
  readonly name = 'acme-notify'

  register(observer: NotificationWatcherObserver, options = { capturePayload: false }) {
const onDelivery = (delivery: AcmeDelivery) => {
  try {
    observer.sent({
      adapter: this.name,
      channel: delivery.channel,
      notification: delivery.template,
      notifiable: delivery.recipientId,
      durationMs: delivery.durationMs,
      ...(options.capturePayload ? { payload: delivery.payload } : {}),
    })
  } catch {}
}
const onFailure = (delivery: AcmeDelivery, error: unknown) => {
  try {
    observer.failed({
      adapter: this.name,
      channel: delivery.channel,
      notification: delivery.template,
      notifiable: delivery.recipientId,
      error,
    })
  } catch {}
}

acmeNotifications.on('sent', onDelivery)
acmeNotifications.on('failed', onFailure)
return () => {
  acmeNotifications.off('sent', onDelivery)
  acmeNotifications.off('failed', onFailure)
}
  }
}
```

Register it under `watchers.notification.adapters` and enable the watcher:

```ts
watchers: {
  notification: {
enabled: true,
adapters: [new AcmeNotificationAdapter()],
capturePayload: false,
  },
},
```

## Socket adapters

`SocketWatcherAdapter` covers connection lifecycle and inbound or outbound messages:

```ts
interface SocketWatcherAdapter {
  readonly name: string
  register(
observer: SocketWatcherObserver,
options?: { capturePayload: boolean }
  ): void | (() => void | Promise<void>) | Promise<void | (() => void | Promise<void>)>
}
```

`connected` requires `adapter` and `socketId`; it may include `transport`, `channel`,
`remoteAddress`, and scalar `userId`. `disconnected` can add `reason` and `durationMs`. `message`
requires `direction: 'inbound' | 'outbound'` and can add `event`, `payload`, and `sizeBytes`.

```ts
export class AcmeSocketAdapter implements SocketWatcherAdapter {
  readonly name = 'acme-socket'

  register(observer: SocketWatcherObserver, options = { capturePayload: false }) {
const onMessage = (socket: AcmeSocket, message: AcmeMessage) => {
  try {
    observer.message({
      adapter: this.name,
      socketId: socket.id,
      transport: 'ws',
      channel: message.channel,
      userId: socket.userId,
      direction: message.direction,
      event: message.event,
      sizeBytes: message.byteLength,
      ...(options.capturePayload ? { payload: message.payload } : {}),
    })
  } catch {}
}

const removeMessage = acmeSockets.onMessage(onMessage)
const removeConnected = acmeSockets.onConnected((socket) => {
  try {
    observer.connected({ adapter: this.name, socketId: socket.id, transport: 'ws' })
  } catch {}
})
const removeDisconnected = acmeSockets.onDisconnected((socket, reason, durationMs) => {
  try {
    observer.disconnected({
      adapter: this.name,
      socketId: socket.id,
      transport: 'ws',
      reason,
      durationMs,
    })
  } catch {}
})

return () => {
  removeMessage()
  removeConnected()
  removeDisconnected()
}
  }
}
```

Register it under `watchers.socket.adapters` with the watcher enabled. The separate Transmit
watcher covers server-to-client broadcasts; this seam covers connection state and traffic Periscope
cannot otherwise observe.

## Custom watcher

A watcher owns a subscription lifecycle:

```ts
interface Watcher {
  readonly name: string
  register(): void | Promise<void>
  cleanup?(): void | Promise<void>
}

type PeriscopeWatcherFactory = (context: WatcherContext) => Watcher
```

Factories receive the AdonisJS application, recorder, and resolved Periscope config. Custom
watchers register after built-ins and clean up first in reverse order. `register` and `cleanup` must
be idempotent; guard host callbacks even though the registry also safeguards lifecycle calls.

```ts
import {
  EntryType,
  IncomingEntry,
  type PeriscopeWatcherFactory,
} from '@rikology/adonisjs-periscope'

const paymentWatcher: PeriscopeWatcherFactory = ({ recorder }) => {
  let unsubscribe: (() => void) | null = null

  return {
name: 'payment',
register() {
  if (unsubscribe !== null) return
  unsubscribe = subscribeToPayments((paymentId) => {
    try {
      recorder.record(
        IncomingEntry.make(EntryType.EVENT, {
          name: 'payment:settled',
          payload: { paymentId },
          isClassEvent: false,
        }).withTags(['domain:payments'])
      )
    } catch {
      // The payment path must not depend on diagnostics.
    }
  })
},
cleanup() {
  unsubscribe?.()
  unsubscribe = null
},
  }
}

export default defineConfig({
  watchers: { custom: [paymentWatcher] },
})
```

Prefer the built-in event watcher when the source is already an AdonisJS emitter event.

## Custom store

`PeriscopeStore` is the complete persistence boundary. A custom driver must implement every
required method:

| Method                                      | Contract                                                                         |
| ------------------------------------------- | -------------------------------------------------------------------------------- |
| `save(entries)`                             | Persist one flushed batch; duplicate UUIDs must not reject the remainder         |
| `find(uuid)`                                | Return one entry or `null`                                                       |
| `list(query?)`                              | Cursor-paginated entries honoring all optional AND filters and sequence ordering |
| `batch(batchId)`                            | Batch entries ordered by sequence ascending                                      |
| `counts(application?)`                      | Counts by entry type                                                             |
| `requestStats(query)`                       | Bounded request buckets, error counts, p50/p95, and optional route groups        |
| `applications()`                            | Application summaries, newest activity first                                     |
| `exceptionGroups(query?)`                   | Cursor-paginated exception families                                              |
| `prune(options)`                            | Delete by default/per-type cutoff, exception policy, and optional application    |
| `trim(maxEntries)`                          | Remove the oldest entries down to the ceiling                                    |
| `clear(application?)`                       | Delete recordings but preserve monitored tags and flags                          |
| `monitoredTags`/`monitorTag`/`unmonitorTag` | Read and mutate exact tags scoped by application                                 |
| flag methods                                | Read, prefix-query, set with optional expiry, and delete flags                   |
| `close()`                                   | Release resources after the provider's final flush                               |
| `diagnostics?()`                            | Optional synchronous write-queue counters for `/api/status`                      |

Methods must be concurrency-safe. Missing rows resolve to `null` or a no-op rather than throwing;
genuine I/O errors may reject because the recorder catches and reports them. Use the shared storage
contract and shipped stores as behavioral references.

A factory receives `{ app, config }` and may be asynchronous:

```ts
import type {
  PeriscopeStore,
  PeriscopeStoreFactory,
  StoredEntry,
} from '@rikology/adonisjs-periscope'

class RedisPeriscopeStore implements PeriscopeStore {
  constructor(
private client: RedisClient,
private applicationName: string
  ) {}

  async save(entries: StoredEntry[]) {
await this.client.saveBatch(entries) // Atomic, bounded, duplicate-safe.
  }

  async find(uuid: string) {
return this.client.findEntry(uuid)
  }

  // Implement list, batch, counts, requestStats, applications, exceptionGroups,
  // prune, trim, clear, monitored-tag operations, and flag operations here.

  async close() {
await this.client.quit()
  }
}

const storeFactory: PeriscopeStoreFactory = async ({ app, config }) => {
  const client = await connectRedis(app.config.get('redis.periscope'))
  return new RedisPeriscopeStore(client, config.applicationName)
}

export default defineConfig({
  storage: {
driver: 'custom',
factory: storeFactory,
  },
})
```

Ace maintenance commands treat a custom store as durable and call it like the built-in SQL drivers.

## Flush fanout

`FlushFanout` connects recorder flushes to SSE subscribers. The default implementation is local to
one process; use this seam for Redis, Transmit, or another pub/sub transport:

```ts
interface FlushFanout {
  publish(event: FlushedEvent): void | Promise<void>
  subscribe(listener: (event: FlushedEvent) => void): () => void
  close?(): void | Promise<void>
}

type FlushFanoutFactory = (
  context: PeriscopeStoreFactoryContext
) => FlushFanout | Promise<FlushFanout>
```

`FlushedEvent` is JSON-safe and content-free. It carries `type`, `uuid`, and an `indexRow` containing
the entry UUID, batch ID, application, type, family hash, tags, index visibility, string sequence,
and ISO creation time. Deliver each event once to every local listener. `subscribe` must return an
idempotent unsubscribe function, listener failures must not starve later listeners, and `close`
must release pub/sub resources.

```ts
import type { FlushFanout, FlushFanoutFactory, FlushedEvent } from '@rikology/adonisjs-periscope'

class RedisFanout implements FlushFanout {
  readonly listeners = new Set<(event: FlushedEvent) => void>()

  constructor(
private publisher: RedisClient,
private subscriber: RedisClient
  ) {}

  async start() {
await this.subscriber.subscribe('periscope:flush', (json) => {
  const event = JSON.parse(json) as FlushedEvent
  for (const listener of [...this.listeners]) {
    try {
      listener(event)
    } catch {}
  }
})
  }

  async publish(event: FlushedEvent) {
await this.publisher.publish('periscope:flush', JSON.stringify(event))
  }

  subscribe(listener: (event: FlushedEvent) => void) {
this.listeners.add(listener)
let active = true
return () => {
  if (!active) return
  active = false
  this.listeners.delete(listener)
}
  }

  async close() {
this.listeners.clear()
await Promise.allSettled([this.subscriber.quit(), this.publisher.quit()])
  }
}

const fanout: FlushFanoutFactory = async ({ app }) => {
  const publisher = await connectRedis(app.config.get('redis.periscope'))
  const subscriber = publisher.duplicate()
  const adapter = new RedisFanout(publisher, subscriber)
  await adapter.start()
  return adapter
}

export default defineConfig({
  dashboard: { fanout },
})
```

The example assumes the pub/sub service echoes publications to the publishing worker. If it does
not, deliver locally in `publish` as well, while preventing duplicates on services that do echo.

## Filter and tag hooks

Hooks are lighter than a watcher when the required entry already exists. They run before buffering:

```ts
export default defineConfig({
  hooks: {
filter: [(entry) => entry.type !== 'request' || entry.content.routePattern !== '/health'],
tag: [
  (entry) =>
    entry.type === 'request' && typeof entry.content.routePattern === 'string'
      ? [`route:${entry.content.routePattern}`]
      : [],
],
  },
})
```

A filter returning `false` drops the entry. A tag hook returns extra exact-match tags. Hook failures
are safeguarded and cannot break the host request.

Source: https://adonisjs-periscope.pages.dev/reference/adapters/index.mdx