bound the intake batch by the queue it is drawn from
204 unmodified linesExpand all20520620720820920921021121221362 unmodified lines204 unmodified linesExpand all webhookEventsChannel: channel(env.WEBHOOK_EVENTS_CHANNEL), webhookCacheMax: integer('WEBHOOK_CACHE_MAX', env.WEBHOOK_CACHE_MAX, 1_000, 1, 10_000), webhookCacheTtlMs: integer('WEBHOOK_CACHE_TTL_MS', env.WEBHOOK_CACHE_TTL_MS, 60_000, 1_000, 3_600_000), intakeQueueMax: integer('WEBHOOK_INTAKE_QUEUE_MAX', env.WEBHOOK_INTAKE_QUEUE_MAX, 512, 1, 10_000), intakeBatchMax: integer('WEBHOOK_INTAKE_BATCH_MAX', env.WEBHOOK_INTAKE_BATCH_MAX, 128, 1, 1_000), intakeBatchMax: integer('WEBHOOK_INTAKE_BATCH_MAX', env.WEBHOOK_INTAKE_BATCH_MAX, 512, 1, 10_000), intakeEventMaxBytes: integer( 'WEBHOOK_INTAKE_EVENT_MAX_BYTES', env.WEBHOOK_INTAKE_EVENT_MAX_BYTES, 1_048_576,62 unmodified lines243 unmodified linesExpand all244245246247248248249250251252253396 unmodified linesExpand all243 unmodified linesExpand all throw new Error('Unsafe Jetstream relay URL') } this.lastAckedCursor = opts.cursor this.maxQueue = opts.maxQueue ?? config.intakeQueueMax this.batchMax = opts.batchMax ?? config.intakeBatchMax // A batch is drawn from the queue, so it can never usefully exceed it. this.batchMax = Math.min(opts.batchMax ?? config.intakeBatchMax, this.maxQueue) this.maxEventBytes = opts.maxEventBytes ?? config.intakeEventMaxBytes this.reconnectMinMs = opts.reconnectMinMs ?? config.jetstreamReconnectMinMs this.reconnectMaxMs = opts.reconnectMaxMs ?? config.jetstreamReconnectMaxMs this.reconnectMaxExponent = opts.reconnectMaxExponent ?? config.jetstreamReconnectMaxExponent396 unmodified linesExpand all