@lastshotlabs/slingshot-kafka

npm install @lastshotlabs/slingshot-kafka

Functions

`createInMemoryDedupStore`

Build a default in-memory LRU MessageDedupStore with TTL eviction.

Each set() records a timestamp and the access order is refreshed by has() lookups. Stale entries past their TTL are treated as misses; cold entries are evicted when the cache exceeds maxKeys.

function createInMemoryDedupStore(options: { maxKeys?: number } = {}): MessageDedupStore

Source: packages/slingshot-kafka/src/kafkaConnectors.ts

`createKafkaAdapter`

Create a Slingshot event bus backed by Kafka durable topics.

Non-durable listeners still execute in-process. Durable listeners are bridged through Kafka topics, consumer groups, and a reconnect buffer for transient producer failures.

function createKafkaAdapter(rawOpts: KafkaAdapterOptions & EventBusSerializationOptions & { /** * Optional metrics sink. When provided, the adapter records publish / * consume / dlq counters, publish/consume durations, and pending-buffer * + connection-state gauges so operators can wire ad-hoc dashboards * without log scraping. Defaults to a no-op emitter. */ metrics?: MetricsEmitter; /** * Optional structured logger. Defaults to a console-backed JSON * logger when omitted. All warn/error paths route through the logger * so no structured information is lost to console formatting. */ logger?: Logger; },): AcknowledgedEventBus &

Source: packages/slingshot-kafka/src/kafkaAdapter.ts

`createKafkaConnectors`

Create a programmatic bridge between the Slingshot event bus and Kafka topics.

Inbound connectors consume Kafka messages into handlers. Outbound connectors subscribe to Slingshot events and publish them to Kafka with buffering and duplicate-produce safeguards.

function createKafkaConnectors(rawOpts: KafkaConnectorsConfig & { /** * Optional metrics sink. When provided, the connector records publish / * consume / dlq counters and durations alongside the existing * observability hooks. Defaults to a no-op emitter. */ metrics?: MetricsEmitter; },): KafkaConnectorHandle

Source: packages/slingshot-kafka/src/kafkaConnectors.ts

`getKafkaAdapterIntrospectionOrNull`

Read Kafka adapter introspection metadata from a bus when available.

function getKafkaAdapterIntrospectionOrNull(bus: SlingshotEventBus,): KafkaAdapterIntrospection | null

Source: packages/slingshot-kafka/src/kafkaAdapter.ts

`kafkaAdapterOptionsSchema`

Zod schema for the programmatic Kafka event-bus adapter configuration.

Config Fields

Field	Description
`autoCreateTopics`	Whether to automatically create topics that do not yet exist on the broker
`brokers`	List of Kafka broker addresses to connect to
`clientId`	Kafka client identifier for this adapter instance
`compression`	Compression codec applied to produced messages (e.g. gzip, snappy, lz4)
`connectionTimeout`	Milliseconds to wait for the initial broker connection before timing out
`defaultPartitions`	Default number of partitions when auto-creating topics
`groupPrefix`	Prefix prepended to all consumer group IDs created by this adapter
`heartbeatInterval`	Interval in milliseconds between consumer heartbeats to the group coordinator
`maxRetries`	Maximum number of retries for failed produce or consume operations
`replicationFactor`	Replication factor used when auto-creating topics
`requestTimeout`	Milliseconds to wait for a broker request response before timing out
`sasl`	SASL authentication configuration for the Kafka connection
`sessionTimeout`	Consumer session timeout in milliseconds; triggers a rebalance if exceeded
`ssl`	TLS/SSL configuration for the Kafka connection
`startFromBeginning`	Whether new consumer groups start reading from the earliest available offset
`topicPrefix`	Prefix prepended to all topic names produced or consumed by this adapter
`validation`	Event payload validation mode: strict rejects, warn logs, off skips validation

Source: packages/slingshot-kafka/src/kafkaAdapter.ts

`kafkaConnectorsSchema`

Zod schema for the programmatic Kafka connector bridge configuration.

Config Fields

Field	Description
`/** Override the default inbound dedup TTL (1h). Set to 0 to disable dedup. */
dedupTtlMs`	Override the default inbound dedup TTL (1h); set to 0 to disable dedup
`brokers`	List of Kafka broker addresses for the connector bridge
`clientId`	Kafka client identifier for this connector bridge instance
`compression`	Default compression codec applied to produced messages across all connectors
`drainIntervalMs`	Interval in milliseconds between pending buffer drain attempts
`hooks`	Observability hooks for monitoring inbound and outbound connector activity
`inbound`	Array of inbound connector definitions consuming from Kafka topics
`maxProduceAttempts`	Maximum number of produce attempts before a buffered outbound message is dropped
`sasl`	SASL authentication configuration for the connector Kafka connection
`serializer`	Default serializer used for all connectors unless overridden per connector
`ssl`	TLS/SSL configuration for the connector Kafka connection
`validationMode`	Default payload validation mode applied to all connectors unless overridden

Source: packages/slingshot-kafka/src/kafkaConnectors.ts

`toGroupId`

Build a consumer group ID scoped to the fully resolved topic name and subscription name.

function toGroupId(prefix: string, topic: string, name: string): string

Source: packages/slingshot-kafka/src/kafkaTopicNaming.ts

`toTopicName`

Convert a Slingshot event key to a Kafka topic name.

The event’s : separators are replaced with . (preserving the namespace convention) and the prefix is prepended verbatim. Inputs that contain Kafka-illegal characters or consecutive colons are passed through; callers are responsible for keeping event names well-formed (use defineEvent and the namespace:resource.action convention).

function toTopicName(prefix: string, event: string): string

Source: packages/slingshot-kafka/src/kafkaTopicNaming.ts