Revert "Revert "Add per-destination event filtering""

CLAUDE.md

@@ -0,0 +1,106 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Overview
+
+The Ambar event-bus **emulator**: a local stand-in for the hosted Ambar platform. It polls rows out of source databases (PostgreSQL / MySQL / SQL Server / plain files), stores them in a file-backed partitioned queue, and POSTs each row to registered HTTP destinations with per-destination cursors and infinite retry. Consuming apps (e.g. `sumori`, `virtual-agent`) run it via docker-compose as their local event bus.
+
+Written in **Haskell** (GHC2021). No relation to the TypeScript app repos' conventions — this repo follows ordinary Haskell/cabal practice.
+
+## Dev environment (macOS, arm64)
+
+```bash
+# Toolchain — match CI: GHC 9.10.1 (cabal 3.10+ works)
+ghcup install ghc 9.10.1 && ghcup set ghc 9.10.1
+
+# The vendored libs are a git submodule — nothing builds without this:
+git submodule update --init
+
+# Native libraries the Haskell bindings link against:
+brew install libpq mysql-client openssl@3 zstd pcre
+```
+
+Homebrew's libpq/openssl are keg-only, and cabal's `--extra-*-dirs` CLI flags do **not** propagate to dependency configure steps — put the paths in the user-global cabal config (`~/.config/cabal/config`) instead:
+
+```
+extra-include-dirs: /opt/homebrew/opt/<pkg>/include
+extra-lib-dirs: /opt/homebrew/opt/<pkg>/lib
+```
+
+for each of `libpq`, `mysql-client`, `openssl@3`, `zstd`, `pcre`. Also keep `/opt/homebrew/opt/libpq/bin` and `/opt/homebrew/opt/mysql-client/bin` on `PATH` (for `pg_config` / `mysql_config`).
+
+## Commands
+
+```bash
+./utils.sh build          # cabal build lib:emulator
+./utils.sh run            # run the emulator (needs --config FILE)
+./utils.sh test           # cabal run emulator-tests  (see test infra below)
+./utils.sh test -- --match "some description"   # subset by hspec description
+./utils.sh typecheck      # ghcid fast-feedback loop on the library
+./utils.sh bench          # benchmarks
+./utils.sh build-docker   # release image (static Alpine binary, build/Dockerfile.static)
+```
+
+Tests expect live **PostgreSQL and MySQL** servers (CI provisions Postgres 16 + MySQL 9; see `.github/workflows/build.yml` for the exact setup). Config/queue/projector suites are pure and run without databases — use `--match` to scope. SQL Server needs no provisioning: the harness `docker run`s `mcr.microsoft.com/azure-sql-edge` itself (works on Apple Silicon).
+
+### Running the full suite locally on macOS (Docker-backed DBs)
+
+The harnesses shell out to `psql`/`mysql` and self-provision users/databases, with three macOS traps: the Haskell `mysql` library always dials the Unix socket `/tmp/mysql.sock` (host `localhost`), Unix sockets can't cross the macOS↔Docker-VM boundary, and MySQL 9's `caching_sha2_password` rejects cold-cache auth over a bridged socket (client treats a socket as a secure channel and sends cleartext; the server sees insecure TCP and refuses it). Verified recipe (286/286 pass, 2026-07-16):
+
+```bash
+# Throwaway servers — port 5432 must be free (stop any app stack using it first)
+docker run -d --rm --name emu-test-pg -e POSTGRES_HOST_AUTH_METHOD=trust -p 5432:5432 postgres:16
+# mysql:8.0 + native password sidesteps the caching_sha2 cold-cache failure (8.0-only flag; removed in 9)
+docker run -d --rm --name emu-test-mysql -e MYSQL_ALLOW_EMPTY_PASSWORD=yes -p 3306:3306 \
+  mysql:8.0 --default-authentication-plugin=mysql_native_password
+
+# Bridge the socket the Haskell client insists on to the container's TCP port.
+# backlog=511 matters: hspec runs specs in parallel and socat's default backlog of 5
+# overflows into ECONNREFUSED mid-suite.
+socat UNIX-LISTEN:/tmp/mysql.sock,fork,reuseaddr,backlog=511 TCP:127.0.0.1:3306 &
+
+PATH="$HOME/.ghcup/bin:/opt/homebrew/opt/libpq/bin:/opt/homebrew/opt/mysql-client/bin:$PATH" \
+  PGHOST=127.0.0.1 PGUSER=postgres cabal test emulator-tests --test-show-details=direct
+```
+
+## Architecture
+
+```
+source DB ──(polling SELECT, RepeatableRead)──▶ topic (file-backed, partitioned)
+                                                    │  one topic per source, partitioned by partitioningColumn
+                                                    ▼
+                              one Projector per destination (consumer group = destination id)
+                                    │ per-partition consumers, per-destination cursor
+                                    │ optional per-destination filter (skip-and-commit)
+                                    ▼
+                              HTTP POST Message{data_source_id, …, payload: <row JSON>}
+                              (infinite fibonacci-backoff retry, commit after success)
+```
+
+| Concern | Where |
+|---|---|
+| Orchestration (`emulate`) | `src/Ambar/Emulator.hs` |
+| Config schema + YAML parsing | `src/Ambar/Emulator/Config.hs` |
+| Delivery loop, filtering, envelope | `src/Ambar/Emulator/Projector.hs` |
+| Polling connectors | `src/Ambar/Emulator/Connector/{Postgres,MySQL,MicrosoftSQLServer,Poll}.hs` |
+| File-backed queue/topics | `src/Ambar/Emulator/Queue/…` |
+| HTTP transport (auth, retry policy decode) | `src/Ambar/Transport/Http.hs` |
+| Source cursors (`state.json`, saved every 30s) | `src/Ambar/Emulator.hs` |
+
+Key semantics:
+
+- **At-least-once, per-partition ordered** delivery; a destination's consumer commits only after a successful send (or an intentional filter skip).
+- **Per-destination filter** (optional `filter: {column, values}` in config): non-matching records are skipped but committed. Records missing the column, or with a non-string value, are **delivered** (fail-open) — config mistakes must degrade to the pre-filter behaviour, never to silent data loss.
+- The full source row (all configured `columns`) is the `payload`; consumers see it wrapped in the `Message` envelope.
+
+## Consumers of this repo
+
+- Apps reference the released image `docker.io/ambarltd/emulator:vX.Y` in their docker-compose (`event-bus` service) and mount their config at `/opt/emulator/config/config.yaml`.
+- Releases: git tags (`v1.x`) + static Docker image via `./utils.sh build-docker`; CI (`.github/workflows/`) builds and tests on push.
+- The consuming apps' config registers one destination per projection/reaction endpoint. Keep any destination `filter.values` lists in lockstep with what the consumer actually accepts — a stale filter silently starves the consumer (the apps should own a drift check on their side).
+
+## Git
+
+- Feature branches; never commit to `main` or push without explicit approval.
+- `deps/haskell-libs` is a pinned submodule — don't update its ref as a side effect.

README.md

@@ -102,6 +102,17 @@ data_destinations:
       - sqlserver_source
       - file_source
 
+    # Optional. Deliver only records whose `column` value is in `values`;
+    # other records are skipped (the destination's cursor still advances).
+    # Omit to deliver every record. Records missing the column (or with a
+    # non-string value in it) are always delivered — a misconfigured column
+    # name degrades to full delivery, never to silently dropped records.
+    filter:
+      column: event_name
+      values:
+        - OrderCreated
+        - OrderUpdated
+
   # Send data to a file. One entry per line.
   - id: file_destination
     description: my projection 1

emulator.cabal

@@ -112,6 +112,7 @@ test-suite emulator-tests
     main-is: Tests.hs
     other-modules:
         Test.Config
+        Test.Projector
         Test.Queue
         Test.Connector
         Test.Connector.File

src/Ambar/Emulator.hs

@@ -123,6 +123,7 @@ emulate logger_ config env = do
         , p_destinationDescription = d_description dest
         , p_sources = sourceTopics
         , p_transport = transport
+        , p_filter = d_filter dest
         }
 
   withDestination dest act =

src/Ambar/Emulator/Config.hs

@@ -8,6 +8,7 @@ module Ambar.Emulator.Config
   , Source(..)
   , DataDestination(..)
   , Destination(..)
+  , DestinationFilter(..)
   , Port(..)
   )
   where
@@ -26,6 +27,8 @@ import qualified Data.Text as Text
 import Data.Text (Text)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as Map
+import Data.Set (Set)
+import qualified Data.Set as Set
 import qualified Data.Yaml as Yaml
 
 import Ambar.Emulator.Connector.Poll (PollingInterval(..))
@@ -72,6 +75,16 @@ data DataDestination = DataDestination
   , d_sources :: [DataSource]
   , d_description :: Text
   , d_destination :: Destination
+  , d_filter :: Maybe DestinationFilter
+  }
+
+-- | Server-side record filter for a destination. Only records whose
+-- @f_column@ value is one of @f_values@ are delivered; everything else is
+-- skipped (the consumer cursor still advances). A destination without a
+-- filter receives every record, as before.
+data DestinationFilter = DestinationFilter
+  { f_column :: Text
+  , f_values :: Set Text
   }
 
 data Destination
@@ -187,6 +200,7 @@ parseDataDestination sourcesMap = Json.withObject "DataSource" $ \o -> do
           [ "Invalid data destination type: '" <> t <> "'."
           , "Expected one of: http-push, file."
           ]
+    d_filter <- o .:? "filter"
     return DataDestination{..}
     where
     parseHTTPPush o = do
@@ -197,6 +211,14 @@ parseDataDestination sourcesMap = Json.withObject "DataSource" $ \o -> do
 
     parseFile o = DestinationFile <$> (o .: "path")
 
+instance FromJSON DestinationFilter where
+  parseJSON = Json.withObject "DestinationFilter" $ \o -> do
+    f_column <- o .: "column"
+    values <- o .: "values"
+    when (null values) $
+      fail "filter.values must not be empty (omit the filter to deliver everything)"
+    return $ DestinationFilter f_column (Set.fromList values)
+
 parseEnvConfigFile :: FilePath -> IO EnvironmentConfig
 parseEnvConfigFile path = do
   bs <- BS.readFile path

src/Ambar/Emulator/Projector.hs

@@ -1,14 +1,19 @@
 module Ambar.Emulator.Projector
   ( Projection(..)
   , project
+  , filterVerdict
+  , FilterVerdict(..)
   , Message(..)
   , Payload(..)
   ) where
 
 {-| A projector reads messages from multiple queues, applies a filter to the
 stream and submits passing messages to a single data destination.
 
-We do not implement filters for now.
+A destination with no filter receives every record. A destination with a
+filter receives only records whose filter-column value is in the allowed
+set; filtered-out records are committed without being sent, so the
+consumer cursor always advances.
 -}
 
 import qualified Data.Aeson as Json
@@ -21,10 +26,14 @@ import Data.Text (Text)
 import qualified Data.Text.Encoding as Text
 import qualified Data.Text as Text
 import Control.Concurrent.Async (forConcurrently_)
+import Control.Monad (when)
 import Control.Monad.Extra (whileM)
+import Data.IORef (newIORef, atomicModifyIORef')
 import GHC.Generics (Generic)
 
-import Ambar.Emulator.Config (Id(..), DataDestination, DataSource(..), Source(..))
+import qualified Data.Set as Set
+
+import Ambar.Emulator.Config (Id(..), DataDestination, DataSource(..), Source(..), DestinationFilter(..))
 import Ambar.Emulator.Queue.Topic (Topic, ReadError(..), PartitionCount(..))
 import qualified Ambar.Emulator.Queue.Topic as Topic
 import Ambar.Emulator.Connector.MicrosoftSQLServer (SQLServer(..))
@@ -44,6 +53,7 @@ data Projection = Projection
   , p_destinationDescription :: Text
   , p_sources :: [(DataSource, Topic)]
   , p_transport :: Some Transport
+  , p_filter :: Maybe DestinationFilter
   }
 
 -- | A record enriched with more information to send to the client.
@@ -62,31 +72,51 @@ newtype Payload = Payload Json.Value
   deriving newtype (ToJSON, FromJSON)
 
 project :: SimpleLogger -> Projection -> IO ()
-project logger_ Projection{..} =
-  forConcurrently_ p_sources projectSource
+project logger_ Projection{..} = do
+  -- One warning per destination per emulator run when the filter fails
+  -- open (reviewer note on PR #56): a typo'd filter column silently
+  -- restores full delivery, which looks exactly like a normal match.
+  -- Warning on every record would flood the log at full event rate, so
+  -- the first fail-open claims this flag and later ones stay quiet.
+  warnedFailOpen <- newIORef False
+  let warnFailOpenOnce logger reason = do
+        firstWarn <- atomicModifyIORef' warnedFailOpen (\claimed -> (True, not claimed))
+        when firstWarn $
+          logWarn logger $
+            "filter fail-open: " <> reason
+            <> ". Delivering the record; the destination filter is not being applied."
+            <> " Further fail-open warnings for this destination are suppressed."
+  forConcurrently_ p_sources (projectSource warnFailOpenOnce)
   where
-  projectSource (source, topic) =
+  projectSource warnFailOpenOnce (source, topic) =
     -- one consumer per partition
     Topic.withConsumers topic group pcount $ \consumers ->
     forConcurrently_ consumers $ \consumer ->
-    whileM $ consume logger consumer source
+    whileM $ consume warnFailOpenOnce logger consumer source
     where
       PartitionCount pcount = Topic.partitionCount topic
       logger =
         annotate ("src: " <> unId (s_id source)) $
         annotate ("dst: " <> unId  p_destination)
         logger_
 
-  consume logger consumer source = do
+  consume warnFailOpenOnce logger consumer source = do
     r <- Topic.read consumer
     case r of
       Left EndOfPartition -> return False
       Left err -> fatal logger (show err)
       Right (bs, meta) -> do
         record <- decode logger bs
         let logger' = annotate (relevantFields (s_source source) record) logger
-        retrying logger' $ Transport.sendJSON p_transport (toMsg source record)
-        logInfo logger' ("sent." :: Text)
+            send = do
+              retrying logger' $ Transport.sendJSON p_transport (toMsg source record)
+              logInfo logger' ("sent." :: Text)
+        case filterVerdict p_filter record of
+          Matched -> send
+          Skipped -> logInfo logger' ("filtered." :: Text)
+          FailedOpen reason -> do
+            warnFailOpenOnce logger' reason
+            send
         Topic.commit consumer meta
         return True
 
@@ -109,6 +139,33 @@ project logger_ Projection{..} =
       fatal logger $ "decoding error: " <> err <> "\nraw: " <> raw
     Right v -> return v
 
+-- | The filter's decision for a record.
+--
+-- 'FailedOpen' means the record is DELIVERED even though the filter could
+-- not be applied (missing column, null or non-string value, non-object
+-- record): a misconfigured column name must degrade to full delivery (the
+-- pre-filter behaviour), never to silently dropped records. The reason is
+-- carried so the projector can warn that the filter is not doing its job.
+data FilterVerdict
+  = Matched            -- ^ deliver: no filter, or the column value is allowed
+  | Skipped            -- ^ commit without delivering
+  | FailedOpen Text    -- ^ deliver, but the filter could not be applied
+  deriving (Eq, Show)
+
+filterVerdict :: Maybe DestinationFilter -> Payload -> FilterVerdict
+filterVerdict Nothing _ = Matched
+filterVerdict (Just DestinationFilter{..}) (Payload value) =
+  case value of
+    Json.Object o ->
+      case KeyMap.lookup (fromString $ Text.unpack f_column) o of
+        Just (Json.String v)
+          | Set.member v f_values -> Matched
+          | otherwise -> Skipped
+        Just Json.Null -> FailedOpen $ "filter column '" <> f_column <> "' is null"
+        Just _ -> FailedOpen $ "filter column '" <> f_column <> "' has a non-string value"
+        Nothing -> FailedOpen $ "filter column '" <> f_column <> "' is missing from the record"
+    _ -> FailedOpen "record is not a JSON object"
+
 -- | Fields to print when a record is sent.
 relevantFields :: Source -> Payload -> Text
 relevantFields source (Payload value) = renderPretty $