Bots, Queues und CLI

Bot-Updates, Queue-Jobs und CLI-Befehle sind begrenzte asynchrone Operationen. Die Beispiele erstellen einen Scope je Update, Job oder Befehl und nutzen await using, wenn die Funktion die gesamte Operation verantwortet.

Sie verwenden den gemeinsamen Graphen. Vergleiche die Arbeitseinheit, die jede Bibliothek dem Anwendungscode übergibt. examples/_shared/container.ts

Beispiel	Inhalt
telegraf.ts	Scope je Telegraf-Update
grammy.ts	Scope je Grammy-Update
bullmq.ts	Scope je BullMQ-Job
commander.ts	Scope je Commander-Befehl
yargs.ts	Scope je Yargs-Befehl

Telegraf

import { Telegraf, type Context, type MiddlewareFn } from 'telegraf'

import {
  buildRootContainer,
  createRequestScope,
  type RequestContainer
} from '../_shared/container.js'

const root = buildRootContainer()

type BotContext = Context & {
  container: RequestContainer
}

const withContainer: MiddlewareFn<BotContext> = async (ctx, next) => {
  /*
   * Re-use the shared per-request scope shape: `update_id` becomes the
   * requestId, the Telegram user id becomes the userId. This way the
   * shared services (Logger, AuditService) get the same fields they would
   * see in HTTP request context
   */
  await using scope = createRequestScope(root, {
    requestId: String(ctx.update.update_id),
    userId: ctx.from?.id !== undefined ? String(ctx.from.id) : undefined
  })

  ctx.container = scope
  await next()
}

export const bot = new Telegraf<BotContext>(process.env.BOT_TOKEN!)

bot.use(withContainer)
bot.start(async (ctx) => {
  const users = await ctx.container.getAsync('users')
  const profile = await users.profile(String(ctx.from?.id ?? 'anonymous'))
  await ctx.reply(`Hello ${profile.name}`)
})

Datei im Repository: examples/workers-cli/telegraf.ts

Grammy

import { Bot, type Context, type MiddlewareFn } from 'grammy'

import {
  buildRootContainer,
  createRequestScope,
  type RequestContainer
} from '../_shared/container.js'

const root = buildRootContainer()

type BotContext = Context & {
  container: RequestContainer
}

const withContainer: MiddlewareFn<BotContext> = async (ctx, next) => {
  await using scope = createRequestScope(root, {
    requestId: String(ctx.update.update_id),
    userId: ctx.from?.id !== undefined ? String(ctx.from.id) : undefined
  })

  ctx.container = scope
  await next()
}

export const bot = new Bot<BotContext>(process.env.BOT_TOKEN!)

bot.use(withContainer)
bot.command('help', async (ctx) => {
  const users = await ctx.container.getAsync('users')
  const profile = await users.profile(String(ctx.from?.id ?? 'anonymous'))
  await ctx.reply(`Chat profile: ${profile.name}`)
})

Datei im Repository: examples/workers-cli/grammy.ts

BullMQ

import { Worker, type Job } from 'bullmq'

import {
  buildRootContainer,
  createRequestScope
} from '../_shared/container.js'

const root = buildRootContainer()

export const worker = new Worker('email', async (job: Job<{ to: string }>) => {
  /*
   * Reuse the same per-request shape for jobs: jobId → requestId, payload
   * recipient → userId so AuditService records pick up the right user
   */
  await using scope = createRequestScope(root, {
    requestId: job.id ?? `job:${job.name}`,
    userId: job.data.to
  })

  scope.get('audit').record('email.sent', { name: job.name, to: job.data.to })
  // Real implementation would call scope.get('mailer').send(job.data)
})

export async function closeWorker() {
  await worker.close()
  await root.dispose()
}

Datei im Repository: examples/workers-cli/bullmq.ts

Commander

import { Command } from 'commander'

import { buildRootContainer } from '../_shared/container.js'

const root = buildRootContainer()

export const program = new Command()

program
  .command('import-users <file>')
  .option('--dry-run')
  .action(async (file: string, options: { dryRun?: boolean }) => {
    /* A command invocation owns one operation scope */
    await using scope = root.createScope({
      request: { requestId: `cli:import-users:${Date.now()}` }
    })

    scope.get('audit').record('cli.import-users.start', {
      file,
      dryRun: options.dryRun === true
    })

    /*
     * Real implementation would resolve `users` with `getAsync()` and stream
     * records from `file`. The minimal demo logs the operation lifecycle
     */
    scope.get('audit').record('cli.import-users.done', { file })
  })

export async function run(argv: string[] = process.argv) {
  try {
    await program.parseAsync(argv)
  } finally {
    await root.dispose()
  }
}

Datei im Repository: examples/workers-cli/commander.ts

Yargs

import yargs from 'yargs'
import { hideBin } from 'yargs/helpers'

import { buildRootContainer } from '../_shared/container.js'

const root = buildRootContainer()

export const cli = yargs(hideBin(process.argv))
  .command(
    'sync <target>',
    'Sync a target',
    (builder) =>
      builder
        .positional('target', { type: 'string', demandOption: true })
        .option('verbose', { type: 'boolean', default: false }),
    async (argv) => {
      /* Each command invocation owns one operation scope */
      await using scope = root.createScope({
        request: { requestId: `cli:sync:${Date.now()}` }
      })

      scope.get('audit').record('cli.sync', {
        target: argv.target,
        verbose: argv.verbose
      })
    }
  )
  .strict()

export async function run() {
  try {
    await cli.parseAsync()
  } finally {
    await root.dispose()
  }
}

Datei im Repository: examples/workers-cli/yargs.ts