@@ -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