kansoku-client (empty) → 0.2.0.0
raw patch · 6 files changed
+1069/−0 lines, 6 filesdep +aesondep +asyncdep +base
Dependencies added: aeson, async, base, bytestring, containers, generic-lens, hspec, http-client, http-client-tls, http-types, kansoku-api, kansoku-client, lens, process, servant, servant-client, servant-client-core, stm, text, time, uuid, wai, warp
Files
- CHANGELOG.md +22/−0
- example/Main.hs +17/−0
- kansoku-client.cabal +92/−0
- src/Kansoku/Client.hs +350/−0
- src/Kansoku/Client/Metrics.hs +66/−0
- test/Spec.hs +522/−0
+ CHANGELOG.md view
@@ -0,0 +1,22 @@+# Changelog++## 0.2.0.0 — 2026-11-12++### Breaking Changes++- Adopt named API routes, typed responses, RFC 9457 problems, and partial batch rejection accounting.++### New Features++- Provide buffered event ingestion with versioned schemas and multi-signal identifiers.+- Add typed queries for filters, whole-range aggregation, funnels, paths, cohorts, organizations, custom metrics, session selection, and resolved users.+- Fetch the public event schema catalog.++### Bug Fixes++- Retry only transport failures and retryable outages; stop retrying permanent API failures.+- Validate metric parameters against the preserved routed path.++### Other Changes++- Bound the library and test dependencies to kansoku-api ^>=0.2.0.0.
+ example/Main.hs view
@@ -0,0 +1,17 @@+module Main (main) where++import Data.Aeson qualified as Aeson+import Data.Text qualified as Text+import Kansoku.Client+import System.Environment (getEnv)++main :: IO ()+main = do+ credential <- Text.pack <$> getEnv "KANSOKU_API_KEY"+ withClient (defaultClientConfig "http://localhost:8080" credential) $ \client -> do+ trackEvent+ client+ emptyIds {userId = Just "user_42"}+ "user_registered"+ ["userId" Aeson..= ("user_42" :: Text.Text)]+ flush client
+ kansoku-client.cabal view
@@ -0,0 +1,92 @@+cabal-version: 3.0+name: kansoku-client+version: 0.2.0.0+synopsis: Haskell server SDK for Kansoku analytics.+description:+ A lightweight client package for emitting Kansoku analytics events from+ Haskell backend services.++license: BSD-3-Clause+author: Nadeem Bitar+maintainer: nadeem@gmail.com+category: Web+build-type: Simple+extra-source-files: CHANGELOG.md++common common+ ghc-options:+ -Wall+ -Wcompat+ -Wredundant-constraints+ -Werror=missing-fields++ default-language: GHC2024+ default-extensions:+ DeriveAnyClass+ DuplicateRecordFields+ MultilineStrings+ OverloadedLabels+ OverloadedStrings++library+ import: common+ hs-source-dirs: src+ exposed-modules:+ Kansoku.Client+ Kansoku.Client.Metrics++ build-depends:+ aeson,+ async,+ base >=4.21 && <5,+ bytestring,+ containers,+ generic-lens >=2.2 && <2.4,+ http-client,+ http-client-tls,+ http-types,+ kansoku-api ^>=0.2.0.0,+ lens ^>=5.3,+ servant,+ servant-client,+ servant-client-core,+ stm,+ text,+ time,+ uuid,++executable kansoku-client-example+ import: common+ hs-source-dirs: example+ main-is: Main.hs+ build-depends:+ aeson,+ base >=4.21 && <5,+ kansoku-client,+ text,++test-suite kansoku-client-test+ import: common+ ghc-options: -threaded+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ build-depends:+ aeson,+ base >=4.21 && <5,+ bytestring,+ containers,+ generic-lens,+ hspec,+ http-client,+ http-types,+ kansoku-api ^>=0.2.0.0,+ kansoku-client,+ process,+ servant,+ servant-client,+ stm,+ text,+ time,+ wai,+ warp,
+ src/Kansoku/Client.hs view
@@ -0,0 +1,350 @@+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wno-orphans #-}++-- | Haskell server SDK for emitting Kansoku analytics events.+module Kansoku.Client+ ( Client,+ ClientConfig (..),+ ClientError (..),+ EventIds (..),+ alias,+ close,+ defaultClientConfig,+ droppedCount,+ emptyIds,+ fetchSchemaCatalog,+ flush,+ group,+ identify,+ identifyWithVersion,+ identifyOnce,+ identifyOnceWithVersion,+ kansokuBeaconContext,+ kansokuCredential,+ newClient,+ track,+ trackEvent,+ trackEventWithVersion,+ withClient,+ )+where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (Async, async, cancel)+import Control.Concurrent.STM (STM, TBQueue, TVar, atomically, isFullTBQueue, modifyTVar', newTBQueueIO, newTVarIO, readTVarIO, tryReadTBQueue, writeTBQueue)+import Control.Exception (bracket)+import Data.Aeson (Object)+import Data.Aeson qualified as Aeson+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Aeson.Types (Pair)+import Data.Bifunctor (first)+import Data.Foldable (traverse_)+import Data.Generics.Labels ()+import Data.Map.Strict (Map)+import Data.Text qualified as Text+import Data.UUID qualified as UUID+import Data.UUID.V4 qualified as UUID+import Kansoku.Api (SchemaAPI)+import Kansoku.Api.Response (KansokuResult (..), ProblemDetails)+import Kansoku.Auth.Api (BeaconContext, IngestAuth)+import Kansoku.Event (AccountId (..), AnalyticsEvent (..), AnonymousId (..), EventId (..), EventName (..), EventSource (..), OrganizationId (..), SessionId (..), UserId (..), baseEvent)+import Kansoku.Event.Contract (SchemaCatalog)+import Kansoku.Ingest.Api (BatchAck, IngestRoutes)+import Kansoku.Ingest.Api qualified as Ingest+import Kansoku.Prelude+import Network.HTTP.Client qualified as HTTP+import Network.HTTP.Client.TLS qualified as HTTP.TLS+import Network.HTTP.Types.Status (statusCode)+import Servant.API (NamedRoutes)+import Servant.API.Experimental.Auth (AuthProtect)+import Servant.Client (BaseUrl, ClientEnv, ClientM, mkClientEnv, parseBaseUrl, runClientM)+import Servant.Client qualified as ServantClient+import Servant.Client.Core.Auth (AuthClientData, AuthenticatedRequest, mkAuthenticatedRequest)+import Servant.Client.Core.Request qualified as ClientRequest+import Servant.Client.Core.Response qualified as ClientResponse+import Servant.Client.Generic (AsClientT)++data ClientConfig = ClientConfig+ { baseUrl :: !Text,+ credential :: !Text,+ sourceService :: !(Maybe Text),+ manager :: !(Maybe HTTP.Manager),+ batchMaxSize :: !Int,+ batchMaxIntervalMicros :: !Int,+ bufferCapacity :: !Int,+ maxRetries :: !Int,+ retryBaseDelayMicros :: !Int+ }+ deriving stock (Generic)++defaultClientConfig :: Text -> Text -> ClientConfig+defaultClientConfig baseUrl credential =+ ClientConfig+ { baseUrl,+ credential,+ sourceService = Nothing,+ manager = Nothing,+ batchMaxSize = 100,+ batchMaxIntervalMicros = 5000000,+ bufferCapacity = 10000,+ maxRetries = 5,+ retryBaseDelayMicros = 100000+ }++data EventIds = EventIds+ { userId :: !(Maybe Text),+ anonymousId :: !(Maybe Text),+ identifiers :: !(Maybe (Map Text Text)),+ organizationId :: !(Maybe Text),+ accountId :: !(Maybe Text),+ sessionId :: !(Maybe Text)+ }+ deriving stock (Generic, Eq, Show)++emptyIds :: EventIds+emptyIds =+ EventIds+ { userId = Nothing,+ anonymousId = Nothing,+ identifiers = Nothing,+ organizationId = Nothing,+ accountId = Nothing,+ sessionId = Nothing+ }++data Client = Client+ { config :: !ClientConfig,+ env :: !ClientEnv,+ queue :: !(TBQueue AnalyticsEvent),+ worker :: !(Async ()),+ dropped :: !(TVar Int)+ }+ deriving stock (Generic)++data ClientError+ = DeliveryFailed !ServantClient.ClientError+ | DeliveryRejected !ProblemDetails+ deriving stock (Eq, Show)++-- NamedRoutes fields are type-family applications, so record-dot syntax cannot+-- see them. Use qualified selector application at call sites.+type instance AuthClientData (AuthProtect "kansoku-ingest") = Text++type instance AuthClientData (AuthProtect "kansoku-beacon") = ()++kansokuBeaconContext :: AuthenticatedRequest BeaconContext+kansokuBeaconContext = mkAuthenticatedRequest () (const id)++kansokuCredential :: Text -> AuthenticatedRequest IngestAuth+kansokuCredential token =+ mkAuthenticatedRequest token $ \value ->+ ClientRequest.addHeader "Authorization" ("Bearer " <> value)++ingestClient :: IngestRoutes (AsClientT ClientM)+ingestClient = ServantClient.client (Proxy @(NamedRoutes IngestRoutes))++schemaClient :: ClientM SchemaCatalog+schemaClient = ServantClient.client (Proxy @SchemaAPI)++newClient :: ClientConfig -> IO Client+newClient cfg = do+ parsed <- parseConfiguredBaseUrl cfg+ mgr <- maybe (HTTP.newManager HTTP.TLS.tlsManagerSettings) pure (cfg ^. #manager)+ queue <- newTBQueueIO (fromIntegral (max 1 (cfg ^. #bufferCapacity)))+ dropped <- newTVarIO 0+ let env = mkClientEnv mgr parsed+ worker <- async (flushLoop cfg env queue dropped)+ pure Client {config = cfg, env, queue, worker, dropped}++withClient :: ClientConfig -> (Client -> IO a) -> IO a+withClient cfg =+ bracket (newClient cfg) close++close :: Client -> IO ()+close clientHandle = do+ flush clientHandle+ cancel (clientHandle ^. #worker)++flush :: Client -> IO ()+flush clientHandle = do+ events <- atomically (drainQueue maxBound (clientHandle ^. #queue))+ traverse_+ (deliverBatch (clientHandle ^. #config) (clientHandle ^. #env) (clientHandle ^. #dropped))+ (chunksOf (max 1 (clientHandle ^. #config . #batchMaxSize)) events)++-- | Count events lost to buffer overflow, permanent delivery failure, or a+-- server-side per-event batch rejection.+droppedCount :: Client -> IO Int+droppedCount clientHandle =+ readTVarIO (clientHandle ^. #dropped)++-- | Fetch the public, versioned event schema catalog.+fetchSchemaCatalog :: Client -> IO (Either ClientError SchemaCatalog)+fetchSchemaCatalog clientHandle =+ first DeliveryFailed <$> runClientM schemaClient (clientHandle ^. #env)++track :: Client -> AnalyticsEvent -> IO ()+track clientHandle event = do+ written <-+ atomically $ do+ full <- isFullTBQueue (clientHandle ^. #queue)+ if full+ then pure False+ else writeTBQueue (clientHandle ^. #queue) event >> pure True+ unless written (atomically (modifyTVar' (clientHandle ^. #dropped) (+ 1)))++trackEvent :: Client -> EventIds -> Text -> [Pair] -> IO ()+trackEvent clientHandle ids name pairs =+ buildEvent (clientHandle ^. #config) ids name pairs >>= track clientHandle++-- | Track an event pinned to an explicit published schema version.+trackEventWithVersion :: Client -> EventIds -> Text -> Int -> [Pair] -> IO ()+trackEventWithVersion clientHandle ids name version pairs =+ buildEvent (clientHandle ^. #config) ids name pairs+ <&> (#schemaVersion ?~ version)+ >>= track clientHandle++-- | Governed current traits. Explicit v1 remains available for identity-only events.+identify :: Client -> Text -> Maybe Text -> [Pair] -> IO ()+identify clientHandle user anonId = identifyWithVersion clientHandle user anonId 2++identifyWithVersion :: Client -> Text -> Maybe Text -> Int -> [Pair] -> IO ()+identifyWithVersion clientHandle user anonId version traits =+ trackEventWithVersion clientHandle (emptyIds & #userId ?~ user & #anonymousId .~ anonId) "identify" version traits++identifyOnce :: Client -> Text -> Maybe Text -> [Pair] -> IO ()+identifyOnce clientHandle user anonId = identifyOnceWithVersion clientHandle user anonId 1++identifyOnceWithVersion :: Client -> Text -> Maybe Text -> Int -> [Pair] -> IO ()+identifyOnceWithVersion clientHandle user anonId version traits =+ trackEventWithVersion clientHandle (emptyIds & #userId ?~ user & #anonymousId .~ anonId) "identify_once" version traits++group :: Client -> EventIds -> Text -> [Pair] -> IO ()+group clientHandle ids organization pairs =+ buildEvent (clientHandle ^. #config) (ids & #organizationId ?~ organization) "group" pairs+ <&> (#schemaVersion ?~ 2)+ >>= track clientHandle++alias :: Client -> Text -> Text -> [Pair] -> IO ()+alias clientHandle previousId newId pairs =+ buildEvent+ (clientHandle ^. #config)+ (emptyIds & #anonymousId ?~ previousId & #userId ?~ newId)+ "alias"+ (pairs <> ["previousId" Aeson..= previousId, "newId" Aeson..= newId])+ >>= track clientHandle++buildEvent :: ClientConfig -> EventIds -> Text -> [Pair] -> IO AnalyticsEvent+buildEvent cfg ids name pairs = do+ eventId <- newEventId+ now <- getCurrentTime+ pure $+ baseEvent eventId (EventName name) Service now now+ & #sourceService+ .~ (cfg ^. #sourceService)+ & #anonymousId+ .~ (AnonymousId <$> ids ^. #anonymousId)+ & #userId+ .~ (UserId <$> ids ^. #userId)+ & #identifiers+ .~ (ids ^. #identifiers)+ & #organizationId+ .~ (OrganizationId <$> ids ^. #organizationId)+ & #accountId+ .~ (AccountId <$> ids ^. #accountId)+ & #sessionId+ .~ (SessionId <$> ids ^. #sessionId)+ & #properties+ .~ pairsObject pairs++newEventId :: IO EventId+newEventId =+ EventId . UUID.toText <$> UUID.nextRandom++pairsObject :: [Pair] -> Object+pairsObject =+ KeyMap.fromList++parseConfiguredBaseUrl :: ClientConfig -> IO BaseUrl+parseConfiguredBaseUrl cfg =+ parseBaseUrl (Text.unpack (cfg ^. #baseUrl))++flushLoop :: ClientConfig -> ClientEnv -> TBQueue AnalyticsEvent -> TVar Int -> IO ()+flushLoop cfg env queue dropped = do+ threadDelay (max 1 (cfg ^. #batchMaxIntervalMicros))+ events <- atomically (drainQueue (max 1 (cfg ^. #batchMaxSize)) queue)+ deliverBatch cfg env dropped events+ flushLoop cfg env queue dropped++drainQueue :: Int -> TBQueue AnalyticsEvent -> STM [AnalyticsEvent]+drainQueue limit queue =+ go limit []+ where+ go 0 acc = pure (reverse acc)+ go remaining acc = do+ next <- tryReadTBQueue queue+ case next of+ Nothing -> pure (reverse acc)+ Just event -> go (remaining - 1) (event : acc)++deliverBatch :: ClientConfig -> ClientEnv -> TVar Int -> [AnalyticsEvent] -> IO ()+deliverBatch _ _ _ [] = pure ()+deliverBatch cfg env dropped events = do+ result <- deliverWithRetry cfg env events+ case result of+ Right ack ->+ atomically (modifyTVar' dropped (+ length (ack ^. #rejected)))+ Left _ -> atomically (modifyTVar' dropped (+ length events))++deliverWithRetry :: ClientConfig -> ClientEnv -> [AnalyticsEvent] -> IO (Either ClientError BatchAck)+deliverWithRetry cfg env events =+ go 0+ where+ go attempt = do+ result <-+ runClientM+ (Ingest.postBatch ingestClient (kansokuCredential (cfg ^. #credential)) (Ingest.ingestEvent <$> events))+ env+ case result of+ Right (KansokuOk ack) -> pure (Right ack)+ Right (KansokuUnavailable envelope)+ | envelope ^. #retryable && attempt < cfg ^. #maxRetries -> do+ threadDelay (backoff attempt)+ go (attempt + 1)+ | otherwise -> pure (Left (DeliveryRejected envelope))+ Right (KansokuRateLimited envelope retryAfter)+ | envelope ^. #retryable && attempt < cfg ^. #maxRetries -> do+ let retryMicros = fromInteger (min (toInteger (maxBound :: Int)) (max 0 (toInteger retryAfter) * 1000000))+ threadDelay (max retryMicros (backoff attempt))+ go (attempt + 1)+ | otherwise -> pure (Left (DeliveryRejected envelope))+ Right (KansokuBadRequest envelope) -> pure (Left (DeliveryRejected envelope))+ Right (KansokuNotFound envelope) -> pure (Left (DeliveryRejected envelope))+ Right (KansokuTooLarge envelope) -> pure (Left (DeliveryRejected envelope))+ Left err+ | not (retryClientError err) -> pure (Left (DeliveryFailed err))+ | attempt >= cfg ^. #maxRetries -> pure (Left (DeliveryFailed err))+ | otherwise -> do+ threadDelay (backoff attempt)+ go (attempt + 1)++ backoff attempt =+ max 0 (cfg ^. #retryBaseDelayMicros) * (2 ^ attempt)++-- Authentication failures occur before MultiVerb and arrive as ClientError.+-- Only transport failures and retryable server failures merit another attempt.+retryClientError :: ServantClient.ClientError -> Bool+retryClientError = \case+ ServantClient.ConnectionError {} -> True+ ServantClient.FailureResponse _ response ->+ let status = statusCode (ClientResponse.responseStatusCode response)+ in status >= 500 && case Aeson.eitherDecode @ProblemDetails (ClientResponse.responseBody response) of+ Right problem -> problem ^. #retryable+ Left _ -> True+ _ -> False++chunksOf :: Int -> [a] -> [[a]]+chunksOf _ [] = []+chunksOf n values =+ let (prefix, suffix) = splitAt n values+ in prefix : chunksOf n suffix
+ src/Kansoku/Client/Metrics.hs view
@@ -0,0 +1,66 @@+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wno-orphans #-}++-- | Typed servant-client access to Kansoku's experiment metrics.+module Kansoku.Client.Metrics+ ( pathsClient,+ experimentConversionClient,+ experimentFunnelClient,+ experimentsClient,+ kansokuMetricsCredential,+ metricsClient,+ cohortsClient,+ metricDefinitionsClient,+ )+where++import Data.Time (Day)+import Kansoku.Api.Response (KansokuResult)+import Kansoku.Auth.Api (MetricsAuth)+import Kansoku.Cohort.Api (Cohorts)+import Kansoku.Metrics.Api (MetricResult, MetricsRoutes, PathBy, TimeGrain)+import Kansoku.Metrics.Api qualified as Metrics+import Kansoku.Metrics.DefinitionApi (MetricDefinitions)+import Kansoku.Metrics.Filter (MetricFilter)+import Kansoku.Prelude+import Servant.API (NamedRoutes)+import Servant.API.Experimental.Auth (AuthProtect)+import Servant.Client (ClientM)+import Servant.Client qualified as ServantClient+import Servant.Client.Core.Auth (AuthClientData, AuthenticatedRequest, mkAuthenticatedRequest)+import Servant.Client.Core.Request qualified as ClientRequest+import Servant.Client.Generic (AsClientT)++type instance AuthClientData (AuthProtect "kansoku-metrics") = Text++kansokuMetricsCredential :: Text -> AuthenticatedRequest MetricsAuth+kansokuMetricsCredential token =+ mkAuthenticatedRequest token $ \value ->+ ClientRequest.addHeader "Authorization" ("Bearer " <> value)++metricsClient :: MetricsRoutes (AsClientT ClientM)+metricsClient = ServantClient.client (Proxy @(NamedRoutes MetricsRoutes))++experimentsClient :: AuthenticatedRequest MetricsAuth -> Day -> Day -> Maybe Text -> Maybe TimeGrain -> ClientM (KansokuResult MetricResult)+experimentsClient credential fromDay toDay experiment grain =+ Metrics.experiments metricsClient credential fromDay toDay experiment grain Nothing Nothing []++experimentConversionClient :: AuthenticatedRequest MetricsAuth -> Text -> Text -> Day -> Day -> ClientM (KansokuResult MetricResult)+experimentConversionClient credential experiment event fromDay toDay =+ Metrics.experimentConversion metricsClient credential experiment event fromDay toDay Nothing []++experimentFunnelClient :: AuthenticatedRequest MetricsAuth -> Text -> Text -> Day -> Day -> Maybe Int -> ClientM (KansokuResult MetricResult)+experimentFunnelClient credential experiment steps fromDay toDay window =+ Metrics.experimentFunnel metricsClient credential experiment steps fromDay toDay window Nothing Nothing Nothing Nothing [] Nothing []++-- | Management and member routes derived from the served cohort contract.+cohortsClient :: Cohorts (AsClientT ClientM)+cohortsClient = ServantClient.client (Proxy @(NamedRoutes Cohorts))++-- | Paths and page flows, derived directly from the served route.+pathsClient :: AuthenticatedRequest MetricsAuth -> Day -> Day -> Maybe Text -> Maybe Text -> Maybe Int -> Maybe Int -> Maybe Int -> Maybe Text -> Maybe Text -> Maybe PathBy -> [MetricFilter] -> ClientM (KansokuResult MetricResult)+pathsClient = Metrics.paths metricsClient++metricDefinitionsClient :: MetricDefinitions (AsClientT ClientM)+metricDefinitionsClient = ServantClient.client (Proxy @(NamedRoutes MetricDefinitions))
+ test/Spec.hs view
@@ -0,0 +1,522 @@+{-# LANGUAGE TypeApplications #-}++module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.STM (TVar, atomically, modifyTVar', newTVarIO, readTVarIO)+import Control.Monad (forM_)+import Data.Aeson qualified as Aeson+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KeyMap+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.ByteString.Lazy qualified as LBS+import Data.Char (isAlphaNum)+import Data.Foldable (traverse_)+import Data.Generics.Labels ()+import Data.Map.Strict qualified as Map+import Data.Text qualified as Text+import Data.Time.Format (defaultTimeLocale, formatTime)+import GHC.Clock (getMonotonicTime)+import Kansoku.Api.Response (KansokuResult (..), problem)+import Kansoku.Api.Response qualified as Response+import Kansoku.Client+import Kansoku.Client qualified as Client+import Kansoku.Client.Metrics+import Kansoku.Event+import Kansoku.Ingest.Api (BatchAck (..), BeaconAPI, BeaconEnvelope (..), PlainTextJSON, ingestEvent)+import Kansoku.Metrics.Api (MetricResult (..), TimeGrain (Day))+import Kansoku.Metrics.Api qualified as Metrics+import Kansoku.Metrics.Filter qualified as Filter+import Kansoku.Prelude+import Network.HTTP.Client qualified as HTTP+import Network.HTTP.Types (Status, status200, status202, status400, status401, status404, status413, status429, status503, statusCode)+import Network.HTTP.Types.Header (hAuthorization, hContentType)+import Network.Wai (Application, Request, pathInfo, queryString, requestHeaders, requestMethod, responseLBS, strictRequestBody)+import Network.Wai.Handler.Warp qualified as Warp+import Servant.API (JSON)+import Servant.Client (mkClientEnv, parseBaseUrl, runClientM)+import Servant.Client qualified as SC+import System.Environment (lookupEnv)+import System.Process (readProcess)+import Test.Hspec+import Text.Read (readMaybe)++main :: IO ()+main =+ hspec spec++spec :: Spec+spec = describe "Kansoku.Client" $ do+ it "generated beacon clients send both body media through the public context helper without auth headers" $ do+ now <- getCurrentTime+ media <- newTVarIO []+ let event = ingestEvent (baseEvent (EventId "beacon-client") (EventName "custom") Web now now)+ body = BeaconEnvelope "ksk_w_fixture" [event]+ app request respond = do+ pathInfo request `shouldBe` ["v1", "beacon"]+ lookup hAuthorization (requestHeaders request) `shouldBe` Nothing+ lookup "X-Api-Key" (requestHeaders request) `shouldBe` Nothing+ atomically $ modifyTVar' media (<> [fromMaybe "" (lookup hContentType (requestHeaders request))])+ bytes <- strictRequestBody request+ fmap (const True) (Aeson.eitherDecode bytes :: Either String BeaconEnvelope) `shouldBe` Right True+ respond (responseLBS status202 [(hContentType, "application/json")] (Aeson.encode (BatchAck 1 [])))+ Warp.testWithApplication (pure app) $ \port -> do+ manager <- HTTP.newManager HTTP.defaultManagerSettings+ url <- parseBaseUrl ("http://127.0.0.1:" <> show port)+ let env = mkClientEnv manager url+ runClientM (SC.client (Proxy @(BeaconAPI '[PlainTextJSON])) kansokuBeaconContext body) env `shouldReturn` Right (KansokuOk (BatchAck 1 []))+ runClientM (SC.client (Proxy @(BeaconAPI '[JSON])) kansokuBeaconContext body) env `shouldReturn` Right (KansokuOk (BatchAck 1 []))+ fmap (BS.takeWhile (/= 59)) <$> readTVarIO media `shouldReturn` ["text/plain", "application/json"]++ it "trackEvent builds service-source events with identity fields outside properties" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ let ids =+ emptyIds+ & #userId+ ?~ "user-1"+ & #anonymousId+ ?~ "anon-1"+ & #identifiers+ ?~ Map.fromList [("fingerprint", "fp-1"), ("device", "dev-1"), ("email_hmac", "v1:opaque")]+ & #organizationId+ ?~ "org-1"+ & #accountId+ ?~ "acct-1"+ & #sessionId+ ?~ "sess-1"+ withClient (testConfig port) $ \client -> do+ trackEvent client ids "user_registered" ["plan" Aeson..= ("pro" :: Text), "userId" Aeson..= ("property-only" :: Text)]+ flush client+ [event] <- readEvents received+ event ^. #eventName `shouldBe` EventName "user_registered"+ event ^. #source `shouldBe` Service+ event ^. #userId `shouldBe` Just (UserId "user-1")+ event ^. #anonymousId `shouldBe` Just (AnonymousId "anon-1")+ event ^. #identifiers+ `shouldBe` Just (Map.fromList [("fingerprint", "fp-1"), ("device", "dev-1"), ("email_hmac", "v1:opaque")])+ event ^. #organizationId `shouldBe` Just (OrganizationId "org-1")+ event ^. #accountId `shouldBe` Just (AccountId "acct-1")+ event ^. #sessionId `shouldBe` Just (SessionId "sess-1")+ KeyMap.lookup (Key.fromString "plan") (event ^. #properties) `shouldBe` Just (Aeson.String "pro")+ KeyMap.lookup (Key.fromString "userId") (event ^. #properties) `shouldBe` Just (Aeson.String "property-only")++ it "trackEventWithVersion pins the shared wire field" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ withClient (testConfig port) $ \client -> do+ trackEventWithVersion client emptyIds "page_view" 1 ["url" Aeson..= ("https://example.com" :: Text)]+ flush client+ [event] <- readEvents received+ event ^. #schemaVersion `shouldBe` Just 1++ it "pins governed identify helpers while preserving explicit identity-only overrides" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ withClient (testConfig port) $ \client -> do+ identify client "user" (Just "cookie") ["plan" Aeson..= ("pro" :: Text)]+ identifyOnce client "user" (Just "cookie") ["country" Aeson..= ("GB" :: Text)]+ identifyWithVersion client "user" (Just "cookie") 1 []+ flush client+ events <- readEvents received+ [(event ^. #eventName, event ^. #schemaVersion, event ^. #userId) | event <- events]+ `shouldBe` [(EventName "identify", Just 2, Just (UserId "user")), (EventName "identify_once", Just 1, Just (UserId "user")), (EventName "identify", Just 1, Just (UserId "user"))]++ it "pins the group helper to governed v2 while retaining organisation context" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ withClient (testConfig port) $ \client -> do+ Client.group client emptyIds "org" ["seats" Aeson..= (10 :: Int)]+ flush client+ [event] <- readEvents received+ event ^. #schemaVersion `shouldBe` Just 2+ event ^. #organizationId `shouldBe` Just (OrganizationId "org")++ it "round-trips a helper-built event through the shared AnalyticsEvent JSON type" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ withClient (testConfig port) $ \client -> do+ identify client "user-2" (Just "anon-2") ["trait" Aeson..= ("value" :: Text)]+ flush client+ [event] <- readEvents received+ Aeson.decode (Aeson.encode event) `shouldBe` Just event+ event ^. #eventName `shouldBe` EventName "identify"+ event ^. #userId `shouldBe` Just (UserId "user-2")+ event ^. #anonymousId `shouldBe` Just (AnonymousId "anon-2")++ it "flushes batches at batch size" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ withClient ((testConfig port) & #batchMaxSize .~ 100 & #batchMaxIntervalMicros .~ 10000000) $ \client -> do+ traverse_ (\n -> trackEvent client emptyIds ("event_" <> Text.pack (show n)) []) [1 :: Int .. 250]+ flush client+ lengths <- fmap length <$> readBatches received+ lengths `shouldBe` [100, 100, 50]++ it "flushes on interval" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ client <- newClient ((testConfig port) & #batchMaxSize .~ 100 & #batchMaxIntervalMicros .~ 50000)+ trackEvent client emptyIds "interval_event" []+ threadDelay 200_000+ close client+ lengths <- fmap length <$> readBatches received+ lengths `shouldBe` [1]++ it "retries failed batch delivery with the same event id" $+ withRecorder (failNTimes 2) $ \port received attempts -> do+ withClient ((testConfig port) & #maxRetries .~ 3 & #retryBaseDelayMicros .~ 1_000) $ \client -> do+ trackEvent client emptyIds "retry_event" []+ flush client+ readTVarIO attempts `shouldReturn` 3+ batches <- readBatches received+ fmap length batches `shouldBe` [1, 1, 1]+ length (uniqueEventIds (concat batches)) `shouldBe` 1++ it "honors a real 429 Retry-After header and retries the same event ID" $+ withRecorder (\attempts -> do n <- readTVarIO attempts; pure (if n == 1 then status429 else status202)) $ \port received attempts -> do+ started <- getMonotonicTime+ withClient ((testConfig port) & #maxRetries .~ 2 & #retryBaseDelayMicros .~ 1_000) $ \client -> do+ trackEvent client emptyIds "rate_limited_event" []+ flush client+ droppedCount client `shouldReturn` 0+ elapsed <- subtract started <$> getMonotonicTime+ elapsed `shouldSatisfy` (>= 1)+ readTVarIO attempts `shouldReturn` 2+ batches <- readBatches received+ fmap length batches `shouldBe` [1, 1]+ length (uniqueEventIds (concat batches)) `shouldBe` 1++ it "does not retry a non-retryable API rejection" $+ withRecorder (const (pure status400)) $ \port _ attempts -> do+ client <- newClient ((testConfig port) & #maxRetries .~ 3 & #retryBaseDelayMicros .~ 1_000)+ trackEvent client emptyIds "invalid_event" []+ flush client+ readTVarIO attempts `shouldReturn` 1+ droppedCount client `shouldReturn` 1+ close client++ it "does not retry an authentication rejection received before the typed handler" $+ withRecorder (const (pure status401)) $ \port _ attempts -> do+ client <- newClient ((testConfig port) & #maxRetries .~ 3 & #retryBaseDelayMicros .~ 1_000)+ trackEvent client emptyIds "unauthorized_event" []+ flush client+ readTVarIO attempts `shouldReturn` 1+ droppedCount client `shouldReturn` 1+ close client++ it "counts a per-event batch rejection as dropped without retrying" $+ withRecorder (const (pure status202)) $ \port _ attempts -> do+ client <- newClient ((testConfig port) & #maxRetries .~ 3 & #retryBaseDelayMicros .~ 1_000)+ trackEvent client emptyIds "server_rejected_event" []+ flush client+ readTVarIO attempts `shouldReturn` 1+ droppedCount client `shouldReturn` 1+ close client++ it "does not retry an oversized-batch API rejection" $+ withRecorder (const (pure status413)) $ \port _ attempts -> do+ client <- newClient ((testConfig port) & #maxRetries .~ 3 & #retryBaseDelayMicros .~ 1_000)+ trackEvent client emptyIds "oversized_batch" []+ flush client+ readTVarIO attempts `shouldReturn` 1+ droppedCount client `shouldReturn` 1+ close client++ it "drops newest events on buffer overflow" $+ withRecorder (const (pure status202)) $ \port _ _ -> do+ client <- newClient ((testConfig port) & #bufferCapacity .~ 2 & #batchMaxIntervalMicros .~ 10000000)+ trackEvent client emptyIds "overflow_1" []+ trackEvent client emptyIds "overflow_2" []+ trackEvent client emptyIds "overflow_3" []+ droppedCount client `shouldReturn` 1+ close client++ it "derives authenticated experiment metrics clients from the shared route record" $+ Warp.testWithApplication (pure metricsRecorderApp) $ \port -> do+ manager <- HTTP.newManager HTTP.defaultManagerSettings+ base <- parseBaseUrl ("http://127.0.0.1:" <> show port)+ let env = mkClientEnv manager base+ credential = kansokuMetricsCredential "test-service-token"+ fromDay = read "2026-08-01"+ toDay = read "2026-08-08"+ exposureResult <-+ runClientM+ (experimentsClient credential fromDay toDay (Just "checkout_v2") (Just Day))+ env+ case exposureResult of+ Right (KansokuOk result) -> metric result `shouldBe` "experiment_exposures"+ other -> expectationFailure ("unexpected exposure client result: " <> show other)+ conversionResult <-+ runClientM+ (experimentConversionClient credential "checkout_v2" "purchase_completed" fromDay toDay)+ env+ case conversionResult of+ Right (KansokuOk result) -> metric result `shouldBe` "experiment_conversion"+ other -> expectationFailure ("unexpected conversion client result: " <> show other)+ funnelResult <-+ runClientM+ (experimentFunnelClient credential "checkout_v2" "checkout_started,purchase_completed" fromDay toDay (Just 604800))+ env+ case funnelResult of+ Right (KansokuOk result) -> metric result `shouldBe` "experiment_funnel"+ other -> expectationFailure ("unexpected funnel client result: " <> show other)++ it "derives authenticated paths clients with route anchors and CSV controls" $+ Warp.testWithApplication (pure metricsRecorderApp) $ \port -> do+ manager <- HTTP.newManager HTTP.defaultManagerSettings+ base <- parseBaseUrl ("http://127.0.0.1:" <> show port)+ result <-+ runClientM+ (pathsClient (kansokuMetricsCredential "test-service-token") (read "2026-07-01") (read "2026-07-03") (Just "/pricing") Nothing (Just 4) (Just 2) (Just 60) (Just "page_view") (Just "x,y") (Just Metrics.PathRoute) [])+ (mkClientEnv manager base)+ result `shouldBe` Right (KansokuOk MetricResult {metric = "paths", points = []})++ it "serializes optional session containsEvent once with credentials and filters" $ do+ captured <- newTVarIO []+ input <- either (fail . Text.unpack) pure (Filter.mkMetricFilter "channel" Filter.FilterEq (Just (Aeson.String "email")))+ let app request respond = do+ pathInfo request `shouldBe` ["metrics", "sessions"]+ lookup hAuthorization (requestHeaders request) `shouldBe` Just "Bearer test-service-token"+ atomically (modifyTVar' captured (<> [queryString request]))+ respond (responseLBS status200 [(hContentType, "application/json")] "{\"metric\":\"sessions\",\"points\":[]}")+ Warp.testWithApplication (pure app) $ \port -> do+ manager <- HTTP.newManager HTTP.defaultManagerSettings+ base <- parseBaseUrl ("http://127.0.0.1:" <> show port)+ forM_ [Nothing, Just "client_error,page_leave"] $ \names -> do+ result <-+ runClientM+ (Metrics.sessions metricsClient (kansokuMetricsCredential "test-service-token") (read "2026-07-01") (read "2026-07-02") (Just "entry_page") (Just Metrics.NoTimeBucket) [input] (Just Metrics.SessionCount) names (Just False))+ (mkClientEnv manager base)+ result `shouldBe` Right (KansokuOk MetricResult {metric = "sessions", points = []})+ requests <- readTVarIO captured+ [[(key, value) | (key, value) <- parameters, key == "containsEvent"] | parameters <- requests] `shouldBe` [[], [("containsEvent", Just "client_error,page_leave")]]+ forM_ requests $ \parameters -> do+ lookup "measure" parameters `shouldBe` Just (Just "count")+ lookup "includeBots" parameters `shouldBe` Just (Just "false")+ traverse Aeson.eitherDecodeStrict' [value | ("filter", Just value) <- parameters] `shouldBe` Right [input]++ it "sends typed repeatable JSON filters and decodes nullable namespaced groups" $ do+ captured <- newTVarIO []+ let app request respond = do+ atomically (modifyTVar' captured (const (queryString request)))+ respond+ ( responseLBS+ status200+ [("content-type", "application/json")]+ "{\"metric\":\"events\",\"points\":[{\"bucket\":null,\"group\":{\"source\":\"web\",\"prop:plan\":null},\"value\":1}]}"+ )+ input <- either (fail . Text.unpack) pure (Filter.mkMetricFilter "user_id" Filter.FilterEq (Just (Aeson.String "é:|,\"\\")))+ secondInput <- either (fail . Text.unpack) pure (Filter.mkMetricFilter "source" Filter.FilterEq (Just (Aeson.String "web")))+ Warp.testWithApplication (pure app) $ \port -> do+ manager <- HTTP.newManager HTTP.defaultManagerSettings+ base <- parseBaseUrl ("http://127.0.0.1:" <> show port)+ result <-+ runClientM+ (Metrics.events metricsClient (kansokuMetricsCredential "test-service-token") (read "2026-07-01") (read "2026-07-02") (Just "source,prop:plan") (Just Metrics.NoTimeBucket) [input, secondInput])+ (mkClientEnv manager base)+ case result of+ Right (KansokuOk payload) -> case points payload of+ [point] -> Map.lookup "prop:plan" (point ^. #group) `shouldBe` Just Nothing+ other -> expectationFailure (show other)+ other -> expectationFailure (show other)+ parameters <- readTVarIO captured+ lookup "timeGrain" parameters `shouldBe` Just (Just "none")+ let encodedFilters = [bytes | ("filter", Just bytes) <- parameters]+ traverse Aeson.eitherDecodeStrict' encodedFilters `shouldBe` Right [input, secondInput]++ it "decodes permanent metric limit problems through the generated client" $ do+ attempts <- newTVarIO (0 :: Int)+ let app _ respond = do+ atomically (modifyTVar' attempts (+ 1))+ respond (responseLBS status400 [("content-type", "application/problem+json")] (Aeson.encode (problem 400 "metric_result_too_large" "narrow the query")))+ Warp.testWithApplication (pure app) $ \port -> do+ manager <- HTTP.newManager HTTP.defaultManagerSettings+ base <- parseBaseUrl ("http://127.0.0.1:" <> show port)+ result <- runClientM (experimentsClient (kansokuMetricsCredential "test-service-token") (read "2026-07-01") (read "2026-07-02") Nothing Nothing) (mkClientEnv manager base)+ case result of+ Right (KansokuBadRequest rejection) -> do+ Response.code rejection `shouldBe` "metric_result_too_large"+ Response.retryable rejection `shouldBe` False+ other -> expectationFailure (show other)+ readTVarIO attempts `shouldReturn` 1++ it "reserves alias IDs against custom property overrides" $+ withRecorder (const (pure status202)) $ \port received _ -> do+ withClient (testConfig port) $ \client -> do+ alias client "anon" "user" ["previousId" Aeson..= ("spoof" :: Text), "newId" Aeson..= ("spoof" :: Text)]+ flush client+ events <- readEvents received+ case events of+ [event] -> validateEvent event `shouldBe` Right event+ other -> expectationFailure (show other)++ it "fetches the public schema catalog through the shared route type" $+ Warp.testWithApplication (pure schemaRecorderApp) $ \port ->+ withClient (testConfig port) $ \client ->+ fetchSchemaCatalog client `shouldReturn` Right (schemaCatalog defaultRegistry)++ it "skips the live integration test unless KANSOKU_SERVER_URL is set" $ do+ required <- lookupEnv "KANSOKU_REQUIRED_INTEGRATION"+ server <- lookupEnv "KANSOKU_SERVER_URL"+ credential <- lookupEnv "KANSOKU_API_KEY"+ case (server, credential) of+ (Nothing, _) -> if required == Just "1" then expectationFailure "required KANSOKU_SERVER_URL is not set" else pendingWith "KANSOKU_SERVER_URL is not set"+ (_, Nothing) -> if required == Just "1" then expectationFailure "required KANSOKU_API_KEY is not set" else pendingWith "KANSOKU_API_KEY is not set"+ (Just url, Just apiKey) -> do+ now <- getCurrentTime+ let eventName = "kansoku_client_integration_" <> Text.pack (filter isAlphaNum (formatTime defaultTimeLocale "%FT%T%QZ" now))+ withClient ((defaultClientConfig (Text.pack url) (Text.pack apiKey)) & #batchMaxSize .~ 2) $ \client -> do+ trackEvent client emptyIds eventName []+ flush client+ pollClickHouseCount eventName `shouldReturn` 1++testConfig :: Int -> ClientConfig+testConfig port =+ (defaultClientConfig ("http://127.0.0.1:" <> Text.pack (show port)) "test-write-key")+ & #batchMaxIntervalMicros+ .~ 1000000+ & #retryBaseDelayMicros+ .~ 1000++withRecorder :: (TVar Int -> IO Status) -> (Int -> TVar [LBS.ByteString] -> TVar Int -> IO ()) -> IO ()+withRecorder status action = do+ received <- newTVarIO []+ attempts <- newTVarIO 0+ Warp.testWithApplication (pure (recorderApp received attempts status)) $ \port ->+ action port received attempts++recorderApp :: TVar [LBS.ByteString] -> TVar Int -> (TVar Int -> IO Status) -> Application+recorderApp received attempts nextStatus request respond = do+ if lookup hAuthorization (requestHeaders request) /= Just "Bearer test-write-key"+ then respond (responseLBS status401 [("content-type", "application/json")] "{}")+ else do+ atomically (modifyTVar' attempts (+ 1))+ body <- strictRequestBody request+ status <- nextStatus attempts+ when (requestMethod request == "POST" && pathInfo request == ["v1", "batch"]) $+ atomically (modifyTVar' received (<> [body]))+ respond (responseLBS status ([("content-type", if statusCode status >= 400 then "application/problem+json" else "application/json")] <> [("Retry-After", "1") | status == status429]) (responseBodyFor status request body))++responseBodyFor :: Status -> Request -> LBS.ByteString -> LBS.ByteString+responseBodyFor status _request _body+ | status == status401 = Aeson.encode (problem 401 "unauthorized" "missing or unknown credential")+ | status == status400 =+ Aeson.encode (problem 400 "validation_failed" "validation failed")+ | status == status429 = Aeson.encode (problem 429 "rate_limited" "ingest rate exceeded")+ | status == status503 =+ Aeson.encode (problem 503 "queue_unavailable" "analytics queue is unavailable")+ | status == status413 =+ Aeson.encode (problem 413 "batch_too_large" "batch exceeds limit")+responseBodyFor _status request body+ | pathInfo request == ["v1", "events"] = "{\"accepted\":1,\"eventId\":\"test\"}"+ | otherwise =+ case Aeson.decode body :: Maybe [AnalyticsEvent] of+ Just events+ | any ((== EventName "server_rejected_event") . (^. #eventName)) events ->+ Aeson.encode+ ( Aeson.object+ [ "accepted" Aeson..= (length events - 1),+ "rejected"+ Aeson..= [ Aeson.object+ [ "index" Aeson..= (0 :: Int),+ "eventId" Aeson..= Aeson.Null,+ "errors" Aeson..= ["rejected by test" :: Text]+ ]+ ]+ ]+ )+ | otherwise -> Aeson.encode (Aeson.object ["accepted" Aeson..= length events, "rejected" Aeson..= ([] :: [Aeson.Value])])+ Nothing -> "{\"accepted\":0,\"rejected\":[]}"++metricsRecorderApp :: Application+metricsRecorderApp request respond =+ if lookup hAuthorization (requestHeaders request) /= Just "Bearer test-service-token"+ then respond (responseLBS status401 [("content-type", "application/json")] "{}")+ else case pathInfo request of+ ["metrics", "paths"]+ | queryValue "start" request == Just "/pricing",+ queryValue "end" request == Nothing,+ queryValue "by" request == Just "route",+ queryValue "steps" request == Just "4",+ queryValue "top" request == Just "2",+ queryValue "window" request == Just "60",+ queryValue "events" request == Just "page_view",+ queryValue "exclude" request == Just "x,y" ->+ ok "paths"+ ["metrics", "experiments"]+ | queryValue "experiment" request == Just "checkout_v2",+ queryValue "timeGrain" request == Just "day" ->+ ok "experiment_exposures"+ ["metrics", "experiments", "conversion"]+ | queryValue "experiment" request == Just "checkout_v2",+ queryValue "event" request == Just "purchase_completed" ->+ ok "experiment_conversion"+ ["metrics", "experiments", "funnel"]+ | queryValue "experiment" request == Just "checkout_v2",+ queryValue "steps" request == Just "checkout_started,purchase_completed",+ queryValue "window" request == Just "604800" ->+ ok "experiment_funnel"+ _ -> respond (responseLBS status400 [("content-type", "application/json")] (Aeson.encode (problem 400 "bad_request" "unexpected metrics request")))+ where+ ok name =+ respond+ ( responseLBS+ status200+ [("content-type", "application/json")]+ (Aeson.encode MetricResult {metric = name, points = []})+ )++schemaRecorderApp :: Application+schemaRecorderApp request respond+ | requestMethod request == "GET",+ pathInfo request == ["v1", "schemas"] =+ respond+ ( responseLBS+ status200+ [("content-type", "application/json")]+ (Aeson.encode (schemaCatalog defaultRegistry))+ )+ | otherwise =+ respond (responseLBS status404 [("content-type", "application/json")] "{}")++queryValue :: ByteString -> Request -> Maybe ByteString+queryValue name request =+ lookup name (queryString request) >>= id++failNTimes :: Int -> TVar Int -> IO Status+failNTimes n attempts = do+ count <- readTVarIO attempts+ pure (if count <= n then status503 else status202)++readEvents :: TVar [LBS.ByteString] -> IO [AnalyticsEvent]+readEvents received =+ concat <$> readBatches received++readBatches :: TVar [LBS.ByteString] -> IO [[AnalyticsEvent]]+readBatches received = do+ bodies <- readTVarIO received+ pure (fromMaybe [] . Aeson.decode <$> bodies)++uniqueEventIds :: [AnalyticsEvent] -> [EventId]+uniqueEventIds =+ foldr add []+ where+ add event seen+ | event ^. #eventId `elem` seen = seen+ | otherwise = event ^. #eventId : seen++pollClickHouseCount :: Text -> IO Int+pollClickHouseCount eventName =+ go (20 :: Int)+ where+ go 0 = clickHouseCount eventName+ go remaining = do+ count <- clickHouseCount eventName+ if count == 1+ then pure count+ else threadDelay 250000 >> go (remaining - 1)++clickHouseCount :: Text -> IO Int+clickHouseCount eventName = do+ port <- fromMaybe "9000" <$> lookupEnv "CLICKHOUSE_TCP_PORT"+ raw <-+ readProcess+ "clickhouse-client"+ ["--port", port, "--query", "SELECT count(DISTINCT event_id) FROM analytics.events_raw FINAL WHERE event_name = '" <> Text.unpack eventName <> "'"]+ ""+ pure (fromMaybe 0 (readMaybe raw))