diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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. diff --git a/example/Main.hs b/example/Main.hs new file mode 100644 --- /dev/null +++ b/example/Main.hs @@ -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 diff --git a/kansoku-client.cabal b/kansoku-client.cabal new file mode 100644 --- /dev/null +++ b/kansoku-client.cabal @@ -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, diff --git a/src/Kansoku/Client.hs b/src/Kansoku/Client.hs new file mode 100644 --- /dev/null +++ b/src/Kansoku/Client.hs @@ -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 diff --git a/src/Kansoku/Client/Metrics.hs b/src/Kansoku/Client/Metrics.hs new file mode 100644 --- /dev/null +++ b/src/Kansoku/Client/Metrics.hs @@ -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)) diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -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))