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:

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:

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)
    }
  }
}

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:

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.

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)
  }
}

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

Notification adapters

NotificationWatcherAdapter reports delivery outcomes:

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.

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)
    }
  }
}

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

Socket adapters

SocketWatcherAdapter covers connection lifecycle and inbound or outbound messages:

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.

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:

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.

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:

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:

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.

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:

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.