git.iko.soy/rowtype-yoga/purescript-yoga-kafka
on main help sign in
Initial: yoga-kafka
29c3c13Mark
.github/workflows/ci.ymlnew file+9
compare with main
unifiedsplit
@@ -0,0 +1,9 @@
1+name: CI
2+on: [push, pull_request]
3+jobs:
4+ build:
5+ runs-on: ubuntu-latest
6+ steps:
7+ - uses: actions/checkout@v4
8+ - uses: rowtype-yoga/purescript-yoga-actions/.github/actions/setup@main
9+ - run: spago build
.gitignorenew file+3
@@ -0,0 +1,3 @@
1+node_modules/
2+.spago/
3+output*/
package.jsonnew file+12
@@ -0,0 +1,12 @@
1+{
2+ "name": "purescript-yoga-kafka",
3+ "version": "0.1.0",
4+ "description": "PureScript FFI bindings for Apache Kafka",
5+ "license": "MIT",
6+ "repository": {
7+ "type": "git",
8+ "url": "https://github.com/rowtype-yoga/purescript-yoga-backend-stack.git",
9+ "directory": "packages/yoga-kafka"
10+ },
11+ "keywords": ["purescript", "ffi", "bindings", "yoga"]
12+}
spago.yamlnew file+20
@@ -0,0 +1,20 @@
1+workspace:
2+ package_set:
3+ registry: 51.1.0
4+
5+package:
6+ name: yoga-kafka
7+ publish:
8+ version: 0.1.0
9+ license: MIT
10+ location:
11+ githubOwner: rowtype-yoga
12+ githubRepo: purescript-yoga-kafka
13+ dependencies:
14+ - js-promise: ">=1.0.0 <2.0.0"
15+ - js-promise-aff: ">=1.0.0 <2.0.0"
16+ - aff: ">=7.0.0 <8.0.0"
17+ - effect: ">=4.0.0 <5.0.0"
18+ - prelude: ">=6.0.0 <7.0.0"
19+ - arrays: ">=7.0.0 <8.0.0"
20+ - foreign-object: ">=4.0.0 <5.0.0"
src/Yoga/Kafka/Admin.jsnew file+11
@@ -0,0 +1,11 @@
1+export const createAdminImpl = (kafka) => kafka.admin();
2+
3+export const connectAdminImpl = (admin) => admin.connect();
4+
5+export const createTopicsImpl = (admin, opts) => admin.createTopics(opts);
6+
7+export const deleteTopicsImpl = (admin, opts) => admin.deleteTopics(opts);
8+
9+export const listTopicsImpl = (admin) => admin.listTopics();
10+
11+export const disconnectAdminImpl = (admin) => admin.disconnect();
src/Yoga/Kafka/Admin.pursnew file+55
@@ -0,0 +1,55 @@
1+module Yoga.Kafka.Admin where
2+
3+import Prelude
4+
5+import Effect (Effect)
6+import Effect.Aff (Aff)
7+import Effect.Uncurried (EffectFn1, EffectFn2, runEffectFn1, runEffectFn2)
8+import Yoga.Kafka.Kafka (Kafka, PartitionCount, ReplicationFactor, TopicName)
9+import Promise (Promise)
10+import Promise.Aff (toAffE) as Promise
11+
12+-- Opaque Admin type
13+foreign import data Admin :: Type
14+
15+-- Create admin
16+foreign import createAdminImpl :: EffectFn1 Kafka Admin
17+
18+createAdmin :: Kafka -> Effect Admin
19+createAdmin = runEffectFn1 createAdminImpl
20+
21+-- Connect admin
22+foreign import connectAdminImpl :: EffectFn1 Admin (Promise Unit)
23+
24+connect :: Admin -> Aff Unit
25+connect = runEffectFn1 connectAdminImpl >>> Promise.toAffE
26+
27+-- Create topics
28+type TopicConfig =
29+ { topic :: TopicName
30+ , numPartitions :: PartitionCount
31+ , replicationFactor :: ReplicationFactor
32+ }
33+
34+foreign import createTopicsImpl :: EffectFn2 Admin { topics :: Array TopicConfig } (Promise Boolean)
35+
36+createTopics :: { topics :: Array TopicConfig } -> Admin -> Aff Boolean
37+createTopics opts admin = runEffectFn2 createTopicsImpl admin opts # Promise.toAffE
38+
39+-- Delete topics
40+foreign import deleteTopicsImpl :: EffectFn2 Admin { topics :: Array TopicName } (Promise Unit)
41+
42+deleteTopics :: { topics :: Array TopicName } -> Admin -> Aff Unit
43+deleteTopics opts admin = runEffectFn2 deleteTopicsImpl admin opts # Promise.toAffE
44+
45+-- List topics
46+foreign import listTopicsImpl :: EffectFn1 Admin (Promise (Array TopicName))
47+
48+listTopics :: Admin -> Aff (Array TopicName)
49+listTopics = runEffectFn1 listTopicsImpl >>> Promise.toAffE
50+
51+-- Disconnect admin
52+foreign import disconnectAdminImpl :: EffectFn1 Admin (Promise Unit)
53+
54+disconnect :: Admin -> Aff Unit
55+disconnect = runEffectFn1 disconnectAdminImpl >>> Promise.toAffE
src/Yoga/Kafka/Consumer.jsnew file+61
@@ -0,0 +1,61 @@
1+export const createConsumerImpl = (kafka, opts) => kafka.consumer(opts);
2+
3+export const connectConsumerImpl = (consumer) => consumer.connect();
4+
5+export const subscribeImpl = (consumer, opts) => consumer.subscribe(opts);
6+
7+// Transform user options to match KafkaJS expectations
8+export const runImpl = (consumer, opts, transformKey, promiseToAff, affToPromise) => {
9+ const wrappedOpts = { ...opts };
10+
11+ // Handle eachBatch if present - transform the batch and wrap the handler
12+ if (opts.eachBatch) {
13+ const userHandler = opts.eachBatch;
14+
15+ wrappedOpts.eachBatch = async (payload) => {
16+ // 1. Transform message keys in-place (Nullable -> Maybe)
17+ for (const msg of payload.batch.messages) {
18+ msg.key = transformKey(msg.key);
19+ }
20+
21+ // 2. Transform heartbeat in-place: Effect (Promise Unit) -> Aff Unit
22+ payload.heartbeat = promiseToAff(payload.heartbeat);
23+
24+ // 3. Flatten structure - move batch fields to top level and expose all utility functions
25+ const flatPayload = {
26+ topic: payload.batch.topic,
27+ partition: payload.batch.partition,
28+ highWatermark: payload.batch.highWatermark, // Last committed offset for lag calculation
29+ messages: payload.batch.messages,
30+ resolveOffset: (offset) => payload.resolveOffset(offset), // Mark message as processed (fire-and-forget)
31+ heartbeat: payload.heartbeat, // Already converted to Aff Unit
32+ commitOffsetsIfNecessary: () => payload.commitOffsetsIfNecessary(), // Manual offset commit
33+ uncommittedOffsets: () => payload.uncommittedOffsets(), // Get uncommitted offsets
34+ isRunning: () => payload.isRunning(), // Check if running
35+ isStale: () => payload.isStale(), // Check if batch is stale
36+ pause: () => payload.pause() // Pause partition (returns resume function)
37+ };
38+
39+ // 4. Call user handler (Batch -> Aff Unit) and convert result to Promise
40+ const affResult = userHandler(flatPayload);
41+ // affToPromise returns Effect (Promise Unit), call it to run the effect and get Promise
42+ return affToPromise(affResult)();
43+ };
44+ }
45+
46+ // Handle eachMessage if present - wrap the handler
47+ if (opts.eachMessage) {
48+ const userHandler = opts.eachMessage;
49+
50+ wrappedOpts.eachMessage = async (payload) => {
51+ // Call user handler (KafkaMessageFFI -> Aff Unit) and convert result to Promise
52+ const affResult = userHandler(payload);
53+ // affToPromise returns Effect (Promise Unit), call it to run the effect and get Promise
54+ return affToPromise(affResult)();
55+ };
56+ }
57+
58+ return consumer.run(wrappedOpts);
59+};
60+
61+export const disconnectConsumerImpl = (consumer) => consumer.disconnect();
src/Yoga/Kafka/Consumer.pursnew file+117
@@ -0,0 +1,117 @@
1+module Yoga.Kafka.Consumer where
2+
3+import Prelude
4+
5+import Data.Maybe (Maybe)
6+import Data.Nullable (Nullable, toMaybe)
7+import Data.Time.Duration (Milliseconds)
8+import Effect (Effect)
9+import Effect.Aff (Aff)
10+import Effect.Uncurried (EffectFn1, EffectFn2, EffectFn5, runEffectFn1, runEffectFn2, runEffectFn5)
11+import Foreign (Foreign)
12+import Foreign.Object (Object)
13+import Yoga.Kafka.Kafka (ConsumerGroupId, HeaderValue, Kafka, Key, Offset, PartitionId, Timestamp, TopicName, Value)
14+import Prim.Row (class Union)
15+import Promise (Promise)
16+import Promise.Aff (fromAff, toAffE) as Promise
17+
18+-- Opaque Consumer type
19+foreign import data Consumer :: Type
20+
21+-- Create consumer
22+type ConsumerOptionsImpl = (sessionTimeout :: Milliseconds, rebalanceTimeout :: Milliseconds, heartbeatInterval :: Milliseconds)
23+
24+foreign import createConsumerImpl :: forall opts. EffectFn2 Kafka { groupId :: ConsumerGroupId | opts } Consumer
25+
26+createConsumer :: forall opts opts_. Union opts opts_ ConsumerOptionsImpl => { groupId :: ConsumerGroupId | opts } -> Kafka -> Effect Consumer
27+createConsumer opts kafka = runEffectFn2 createConsumerImpl kafka opts
28+
29+-- Connect consumer
30+foreign import connectConsumerImpl :: EffectFn1 Consumer (Promise Unit)
31+
32+connect :: Consumer -> Aff Unit
33+connect = runEffectFn1 connectConsumerImpl >>> Promise.toAffE
34+
35+-- Subscribe to topics
36+type SubscribeOptionsImpl = (fromBeginning :: Boolean)
37+
38+foreign import subscribeImpl :: forall opts. EffectFn2 Consumer { topic :: TopicName | opts } (Promise Unit)
39+
40+subscribe :: forall opts opts_. Union opts opts_ SubscribeOptionsImpl => { topic :: TopicName | opts } -> Consumer -> Aff Unit
41+subscribe opts consumer = runEffectFn2 subscribeImpl consumer opts # Promise.toAffE
42+
43+-- FFI types
44+type KafkaMessageFFI =
45+ { topic :: TopicName
46+ , partition :: PartitionId
47+ , message ::
48+ { key :: Nullable Key
49+ , value :: Value
50+ , headers :: Object HeaderValue
51+ , offset :: Offset
52+ , timestamp :: Timestamp
53+ }
54+ }
55+
56+-- PureScript-friendly batch types (mutated in JavaScript before reaching PureScript)
57+type BatchMessage =
58+ { key :: Maybe Key
59+ , value :: Value
60+ , headers :: Object HeaderValue
61+ , offset :: Offset
62+ , timestamp :: Timestamp
63+ }
64+
65+-- Batch type with async operations converted to Aff (mutated by JavaScript)
66+type Batch =
67+ { topic :: TopicName
68+ , partition :: PartitionId
69+ , highWatermark :: Offset -- Last committed offset, useful for calculating lag
70+ , messages :: Array BatchMessage
71+ , resolveOffset :: Offset -> Effect Unit -- Mark a message as processed (synchronous, returns void)
72+ , heartbeat :: Aff Unit -- Send heartbeat to broker (converted from Effect (Promise Unit))
73+ , commitOffsetsIfNecessary :: Effect (Promise Unit) -- Commit offsets based on autoCommit config
74+ , uncommittedOffsets :: Effect Foreign -- Get uncommitted offsets by topic-partition (complex nested structure)
75+ , isRunning :: Effect Boolean -- Check if consumer is still running
76+ , isStale :: Effect Boolean -- Check if batch is stale (e.g., after seek)
77+ , pause :: Effect (Effect Unit) -- Pause current partition, returns resume function
78+ }
79+
80+-- User-facing run options (PureScript-friendly!)
81+-- Handlers are Aff Unit to support async operations (use liftEffect for Effect code)
82+type RunOptionsImpl =
83+ ( eachMessage :: KafkaMessageFFI -> Aff Unit
84+ , eachBatch :: Batch -> Aff Unit
85+ , eachBatchAutoResolve :: Boolean -- Auto-commit last offset if eachBatch doesn't throw (default: true)
86+ , autoCommit :: Boolean -- Enable/disable auto commit (default: true)
87+ , autoCommitInterval :: Milliseconds -- Commit every N milliseconds
88+ , autoCommitThreshold :: Int -- Commit after N messages
89+ , partitionsConsumedConcurrently :: Int -- Process N partitions concurrently (default: 1)
90+ )
91+
92+foreign import runImpl
93+ :: forall opts
94+ . EffectFn5
95+ Consumer
96+ { | opts }
97+ (Nullable Key -> Maybe Key)
98+ (Effect (Promise Unit) -> Aff Unit)
99+ (Aff Unit -> Effect (Promise Unit))
100+ (Promise Unit)
101+
102+run :: forall opts opts_. Union opts opts_ RunOptionsImpl => { | opts } -> Consumer -> Aff Unit
103+run opts consumer =
104+ runEffectFn5
105+ runImpl
106+ consumer
107+ opts
108+ toMaybe
109+ Promise.toAffE
110+ (\aff -> Promise.fromAff aff)
111+ # Promise.toAffE
112+
113+-- Disconnect consumer
114+foreign import disconnectConsumerImpl :: EffectFn1 Consumer (Promise Unit)
115+
116+disconnect :: Consumer -> Aff Unit
117+disconnect = runEffectFn1 disconnectConsumerImpl >>> Promise.toAffE
src/Yoga/Kafka/Kafka.jsnew file+3
@@ -0,0 +1,3 @@
1+import { Kafka } from 'kafkajs';
2+
3+export const kafkaImpl = (config) => new Kafka(config);
src/Yoga/Kafka/Kafka.pursnew file+110
@@ -0,0 +1,110 @@
1+module Yoga.Kafka.Kafka where
2+
3+import Prelude
4+
5+import Data.Newtype (class Newtype)
6+import Effect (Effect)
7+import Effect.Uncurried (EffectFn1, runEffectFn1)
8+import Prim.Row (class Union)
9+
10+-- Opaque Kafka client type
11+foreign import data Kafka :: Type
12+
13+-- Newtypes for type safety
14+newtype TopicName = TopicName String
15+
16+derive instance Newtype TopicName _
17+derive newtype instance Eq TopicName
18+derive newtype instance Ord TopicName
19+derive newtype instance Show TopicName
20+
21+newtype ConsumerGroupId = ConsumerGroupId String
22+
23+derive instance Newtype ConsumerGroupId _
24+derive newtype instance Eq ConsumerGroupId
25+derive newtype instance Show ConsumerGroupId
26+
27+newtype ClientId = ClientId String
28+
29+derive instance Newtype ClientId _
30+derive newtype instance Eq ClientId
31+derive newtype instance Show ClientId
32+
33+newtype BrokerAddress = BrokerAddress String
34+
35+derive instance Newtype BrokerAddress _
36+derive newtype instance Eq BrokerAddress
37+derive newtype instance Show BrokerAddress
38+
39+newtype PartitionId = PartitionId Int
40+
41+derive instance Newtype PartitionId _
42+derive newtype instance Eq PartitionId
43+derive newtype instance Ord PartitionId
44+derive newtype instance Show PartitionId
45+
46+newtype BrokerId = BrokerId String
47+
48+derive instance Newtype BrokerId _
49+derive newtype instance Eq BrokerId
50+derive newtype instance Show BrokerId
51+
52+newtype Offset = Offset String
53+
54+derive instance Newtype Offset _
55+derive newtype instance Eq Offset
56+derive newtype instance Ord Offset
57+derive newtype instance Show Offset
58+
59+newtype Timestamp = Timestamp String
60+
61+derive instance Newtype Timestamp _
62+derive newtype instance Eq Timestamp
63+derive newtype instance Ord Timestamp
64+derive newtype instance Show Timestamp
65+
66+newtype Key = Key String
67+
68+derive instance Newtype Key _
69+derive newtype instance Eq Key
70+derive newtype instance Show Key
71+
72+newtype Value = Value String
73+
74+derive instance Newtype Value _
75+derive newtype instance Eq Value
76+derive newtype instance Show Value
77+
78+newtype HeaderValue = HeaderValue String
79+
80+derive instance Newtype HeaderValue _
81+derive newtype instance Eq HeaderValue
82+derive newtype instance Show HeaderValue
83+
84+newtype PartitionCount = PartitionCount Int
85+
86+derive instance Newtype PartitionCount _
87+derive newtype instance Eq PartitionCount
88+derive newtype instance Ord PartitionCount
89+derive newtype instance Show PartitionCount
90+
91+newtype ReplicationFactor = ReplicationFactor Int
92+
93+derive instance Newtype ReplicationFactor _
94+derive newtype instance Eq ReplicationFactor
95+derive newtype instance Ord ReplicationFactor
96+derive newtype instance Show ReplicationFactor
97+
98+newtype ErrorCode = ErrorCode Int
99+
100+derive instance Newtype ErrorCode _
101+derive newtype instance Eq ErrorCode
102+derive newtype instance Show ErrorCode
103+
104+-- Create Kafka client
105+type KafkaConfigImpl = (brokers :: Array BrokerAddress, clientId :: ClientId)
106+
107+foreign import kafkaImpl :: forall opts. EffectFn1 { | opts } Kafka
108+
109+createKafka :: forall opts opts_. Union opts opts_ KafkaConfigImpl => { | opts } -> Effect Kafka
110+createKafka opts = runEffectFn1 kafkaImpl opts
src/Yoga/Kafka/Producer.jsnew file+7
@@ -0,0 +1,7 @@
1+export const createProducerImpl = (kafka, opts) => kafka.producer(opts);
2+
3+export const connectProducerImpl = (producer) => producer.connect();
4+
5+export const sendImpl = (producer, record) => producer.send(record);
6+
7+export const disconnectProducerImpl = (producer) => producer.disconnect();
src/Yoga/Kafka/Producer.pursnew file+55
@@ -0,0 +1,55 @@
1+module Yoga.Kafka.Producer where
2+
3+import Prelude
4+
5+import Data.Nullable (Nullable)
6+import Data.Time.Duration (Milliseconds)
7+import Effect (Effect)
8+import Effect.Aff (Aff)
9+import Effect.Uncurried (EffectFn1, EffectFn2, runEffectFn1, runEffectFn2)
10+import Foreign.Object (Object)
11+import Yoga.Kafka.Kafka (ErrorCode, HeaderValue, Kafka, Key, Offset, PartitionId, Timestamp, TopicName, Value)
12+import Prim.Row (class Union)
13+import Promise (Promise)
14+import Promise.Aff (toAffE) as Promise
15+
16+-- Opaque Producer type
17+foreign import data Producer :: Type
18+
19+-- Create producer
20+type ProducerOptionsImpl = (allowAutoTopicCreation :: Boolean, transactionTimeout :: Milliseconds)
21+
22+foreign import createProducerImpl :: forall opts. EffectFn2 Kafka { | opts } Producer
23+
24+createProducer :: forall opts opts_. Union opts opts_ ProducerOptionsImpl => { | opts } -> Kafka -> Effect Producer
25+createProducer opts kafka = runEffectFn2 createProducerImpl kafka opts
26+
27+-- Connect producer
28+foreign import connectProducerImpl :: EffectFn1 Producer (Promise Unit)
29+
30+connect :: Producer -> Aff Unit
31+connect = runEffectFn1 connectProducerImpl >>> Promise.toAffE
32+
33+-- Send messages
34+type ProducerMessageOptionsImpl = (key :: Key, partition :: PartitionId, headers :: Object HeaderValue, timestamp :: Timestamp)
35+
36+type RecordMetadata =
37+ { topicName :: TopicName
38+ , partition :: PartitionId
39+ , errorCode :: ErrorCode
40+ , offset :: Nullable Offset
41+ , timestamp :: Nullable Timestamp
42+ }
43+
44+type SendRecordImpl = (topic :: TopicName, messages :: Array (Record ProducerMessageOptionsImpl))
45+
46+foreign import sendImpl :: forall opts msg. EffectFn2 Producer { topic :: TopicName, messages :: Array { value :: Value | msg } | opts } (Promise (Array RecordMetadata))
47+
48+send :: forall opts opts_ msg msgOpts_. Union opts opts_ SendRecordImpl => Union msg msgOpts_ ProducerMessageOptionsImpl => { topic :: TopicName, messages :: Array { value :: Value | msg } | opts } -> Producer -> Aff (Array RecordMetadata)
49+send record producer = runEffectFn2 sendImpl producer record # Promise.toAffE
50+
51+-- Disconnect producer
52+foreign import disconnectProducerImpl :: EffectFn1 Producer (Promise Unit)
53+
54+disconnect :: Producer -> Aff Unit
55+disconnect = runEffectFn1 disconnectProducerImpl >>> Promise.toAffE