Server

Tag Filtering

Server-side message filtering using key-value tags to reduce unnecessary client traffic.

Overview

Tag filtering allows clients to subscribe to channels with server-side filters, receiving only messages that match specified criteria. This dramatically reduces bandwidth and client-side processing by filtering messages at the source.

Use cases:

Subscribe to specific stock symbols on a market data channel
Filter IoT sensor data by location or sensor type
Receive only high-priority notifications
Filter gaming events by player region or match type

Benefits:

Bandwidth savings: Clients receive only relevant messages
Reduced client load: No client-side filtering logic needed
Scalability: Server handles filtering once, benefits all matching clients
Security: Clients can't see data they shouldn't access

Quick Start

Enable Tag Filtering

Configure in config/config.json:

config/config.json

{
  "tag_filtering": {
    "enabled": true,
    "enable_tags": true
  }
}

Option	Default	Description
`enabled`	`false`	Allow clients to subscribe with filters
`enable_tags`	`true`	Include tags in messages sent to clients

Publishing with Tags

Add tags when publishing via HTTP API:

curl -X POST http://localhost:6001/apps/app-id/events \
  -H "Content-Type: application/json" \
  -d '{
    "name": "trade",
    "channel": "market-data",
    "data": "{\"symbol\":\"BTC\",\"price\":50000,\"volume\":1.5}",
    "tags": {
      "symbol": "BTC",
      "exchange": "NYSE",
      "type": "crypto"
    }
  }'

Tags are metadata used for routing and filtering. They don't modify the message payload.

Client Subscription with Filters

Using @sockudo/client

@sockudo/client provides a fluent Filter API:

import Pusher from "@sockudo/client";
import { Filter } from "@sockudo/client/filter";

const pusher = new Pusher("app-key", {
  wsHost: "127.0.0.1",
  wsPort: 6001,
  forceTLS: false
});

// Subscribe with a simple equality filter
const channel = pusher.subscribe(
  "market-data",
  Filter.eq("symbol", "BTC")
);

channel.bind("trade", (data) => {
  console.log("BTC trade:", data);
});

Filter Operators

Equality

// Exact match
Filter.eq("symbol", "BTC")
// Matches: {"symbol": "BTC"}
// Doesn't match: {"symbol": "ETH"}

Inequality

// Not equal
Filter.neq("status", "inactive")
// Matches: {"status": "active"}, {"status": "pending"}
// Doesn't match: {"status": "inactive"}

Comparison

// Greater than
Filter.gt("price", "1000")
// Matches: {"price": "1001"}, {"price": "5000"}
// Doesn't match: {"price": "1000"}, {"price": "500"}

// Greater than or equal
Filter.gte("volume", "100")
// Matches: {"volume": "100"}, {"volume": "200"}
// Doesn't match: {"volume": "99"}

// Less than
Filter.lt("quantity", "50")
// Matches: {"quantity": "49"}, {"quantity": "10"}
// Doesn't match: {"quantity": "50"}, {"quantity": "100"}

// Less than or equal
Filter.lte("age", "30")
// Matches: {"age": "30"}, {"age": "25"}
// Doesn't match: {"age": "31"}

Set Membership

// In set
Filter.in("symbol", ["BTC", "ETH", "SOL"])
// Matches: {"symbol": "BTC"}, {"symbol": "ETH"}
// Doesn't match: {"symbol": "DOGE"}

// Not in set
Filter.nin("region", ["US", "EU"])
// Matches: {"region": "ASIA"}, {"region": "LATAM"}
// Doesn't match: {"region": "US"}

Existence

// Tag exists
Filter.exists("urgent")
// Matches: {"urgent": "true"}, {"urgent": "false"}, {"urgent": ""}
// Doesn't match: {} (no urgent tag)

// Tag doesn't exist
Filter.nexists("deprecated")
// Matches: {} (no deprecated tag)
// Doesn't match: {"deprecated": "true"}

Pattern Matching

// Prefix match
Filter.prefix("symbol", "BTC")
// Matches: {"symbol": "BTC"}, {"symbol": "BTCUSD"}
// Doesn't match: {"symbol": "ETH"}

// Suffix match
Filter.suffix("symbol", "USD")
// Matches: {"symbol": "BTCUSD"}, {"symbol": "ETHUSD"}
// Doesn't match: {"symbol": "BTC"}

Combining Filters

AND Logic

All conditions must match:

// Multiple conditions (implicit AND)
Filter.and(
  Filter.eq("exchange", "NYSE"),
  Filter.eq("type", "crypto"),
  Filter.gt("volume", "1000")
)
// Matches only if ALL conditions are true

OR Logic

Any condition can match:

// Any condition matches
Filter.or(
  Filter.eq("symbol", "BTC"),
  Filter.eq("symbol", "ETH"),
  Filter.eq("symbol", "SOL")
)
// Matches if ANY condition is true

Complex Combinations

Nest AND/OR for complex logic:

// (symbol == BTC OR symbol == ETH) AND volume > 1000
Filter.and(
  Filter.or(
    Filter.eq("symbol", "BTC"),
    Filter.eq("symbol", "ETH")
  ),
  Filter.gt("volume", "1000")
)

// High priority OR (urgent AND large volume)
Filter.or(
  Filter.eq("priority", "high"),
  Filter.and(
    Filter.exists("urgent"),
    Filter.gt("volume", "10000")
  )
)

Practical Examples

Market Data Filtering

// Subscribe to BTC trades only
const btcChannel = pusher.subscribe(
  "market-data",
  Filter.eq("symbol", "BTC")
);

// Subscribe to large trades (any symbol)
const largeTradesChannel = pusher.subscribe(
  "market-data",
  Filter.gt("volume", "5000")
);

// Subscribe to crypto on specific exchanges
const cryptoExchangeChannel = pusher.subscribe(
  "market-data",
  Filter.and(
    Filter.eq("type", "crypto"),
    Filter.in("exchange", ["Coinbase", "Binance", "Kraken"])
  )
);

IoT Sensor Filtering

// Temperature sensors in building A
const tempSensors = pusher.subscribe(
  "iot-sensors",
  Filter.and(
    Filter.eq("building", "A"),
    Filter.eq("type", "temperature")
  )
);

// Critical alerts only
const criticalAlerts = pusher.subscribe(
  "iot-sensors",
  Filter.and(
    Filter.eq("level", "critical"),
    Filter.exists("alert")
  )
);

// Sensors in specific zones
const zoneMonitoring = pusher.subscribe(
  "iot-sensors",
  Filter.in("zone", ["Z1", "Z2", "Z3"])
);

Notification Filtering

// High priority notifications
const urgent = pusher.subscribe(
  "notifications",
  Filter.eq("priority", "high")
);

// Notifications for specific user
const userNotifications = pusher.subscribe(
  "notifications",
  Filter.eq("user_id", currentUserId)
);

// Important OR mentions
const importantOrMentions = pusher.subscribe(
  "notifications",
  Filter.or(
    Filter.eq("important", "true"),
    Filter.exists("mention")
  )
);

Gaming Events Filtering

// Events for specific match
const matchEvents = pusher.subscribe(
  "game-events",
  Filter.eq("match_id", "12345")
);

// Events for player's region
const regionalEvents = pusher.subscribe(
  "game-events",
  Filter.and(
    Filter.eq("region", playerRegion),
    Filter.neq("type", "debug")
  )
);

// High-value loot drops
const lootDrops = pusher.subscribe(
  "game-events",
  Filter.and(
    Filter.eq("event_type", "loot_drop"),
    Filter.gte("rarity", "epic")
  )
);

Publishing Patterns

Basic Publishing

With tags for filtering:

curl -X POST http://localhost:6001/apps/app-id/events \
  -H "Content-Type: application/json" \
  -d '{
    "name": "sensor-reading",
    "channel": "iot-sensors",
    "data": "{\"temperature\":72,\"humidity\":45}",
    "tags": {
      "building": "A",
      "floor": "3",
      "type": "temperature",
      "zone": "Z1"
    }
  }'

Without tags (broadcast to all):

curl -X POST http://localhost:6001/apps/app-id/events \
  -H "Content-Type: application/json" \
  -d '{
    "name": "announcement",
    "channel": "global",
    "data": "{\"message\":\"System maintenance in 1 hour\"}"
  }'

Tag Best Practices

✅ Good tag design:

{
  "tags": {
    "entity_type": "user",
    "entity_id": "12345",
    "action": "login",
    "region": "us-east",
    "priority": "normal"
  }
}

❌ Poor tag design:

{
  "tags": {
    "message": "User 12345 logged in",  // Don't duplicate data payload
    "timestamp": "2024-01-15T10:30:00Z",  // Tags should be static/categorical
    "random_uuid": "a1b2c3d4"  // Unique values defeat filtering
  }
}

Tag Design Guidelines

Use categorical values: Tags should represent categories, not unique identifiers in data
- ✅ "region": "us-east"
- ❌ "request_id": "uuid-1234"
Keep tags simple: Use strings for all tag values
- ✅ "priority": "high"
- ❌ "priority": {"level": 5, "urgent": true}
Plan for filtering: Include tags you'll filter on
- ✅ "symbol": "BTC", "type": "trade"
- ❌ Only putting data in payload
Consistent naming: Use same tag names across messages
- ✅ Always "user_id"
- ❌ Sometimes "user_id", sometimes "userId", sometimes "uid"
Don't over-tag: Only include tags useful for filtering
- ✅ 3-7 meaningful tags
- ❌ 50 tags duplicating entire payload

Tag Visibility Control

Hide Tags from Clients

Save bandwidth by omitting tags in messages to clients:

config/config.json

{
  "tag_filtering": {
    "enabled": true,
    "enable_tags": false  // Tags used for routing only, not sent to clients
  }
}

With enable_tags: true:

{
  "event": "trade",
  "data": "{\"price\":50000}",
  "tags": {
    "symbol": "BTC",
    "exchange": "NYSE"
  }
}

With enable_tags: false:

{
  "event": "trade",
  "data": "{\"price\":50000}"
}

Bandwidth saved: ~50-200 bytes per message depending on tag count.

Per-Channel Control

Override tag visibility per channel:

App Config

{
  "channel_delta_compression": {
    "market:*": {
      "enable_tags": false  // Hide tags on market channels
    },
    "debug:*": {
      "enable_tags": true   // Show tags on debug channels
    }
  }
}

Combining with Delta Compression

Use tag filtering and delta compression together:

Client Setup

import Pusher from "@sockudo/client";
import { Filter } from "@sockudo/client/filter";

const pusher = new Pusher("app-key", {
  wsHost: "127.0.0.1",
  wsPort: 6001,
  forceTLS: false,
  deltaCompression: {
    enabled: true,
    algorithms: ["fossil"]
  }
});

// Subscribe with both filter and delta
const channel = pusher.subscribe(
  "market-data",
  Filter.eq("symbol", "BTC"),
  {
    delta: { enabled: true }
  }
);

Server Configuration

App Config

{
  "channel_delta_compression": {
    "market-data": {
      "enabled": true,
      "algorithm": "fossil",
      "conflation_key": "symbol",
      "max_messages_per_key": 100,
      "max_conflation_keys": 1000,
      "enable_tags": false  // Tags for filtering only
    }
  }
}

How they work together:

Server receives message with tags and data
Tags route message to filtered subscribers
Delta compression reduces message size
Client receives compressed, filtered message

Performance Considerations

Server-Side Load

Filtering cost:

Very low: Simple equality checks (eq, neq, exists)
Low: Set membership checks (in, nin)
Moderate: Comparison operators (gt, lt, gte, lte)
Low: Pattern matching (prefix, suffix)

Optimization tips:

Use simple filters when possible
Prefer eq and in over complex combinations
Limit tag count to 5-10 per message
Use consistent tag naming

Bandwidth Savings

Example scenario:

Channel: 1000 messages/second
10 different symbols
Client interested in 1 symbol

Without filtering:

Client receives: 1000 msg/sec × 200 bytes = 200 KB/sec
Client discards: 900 msg/sec (90%)

With filtering:

Client receives: 100 msg/sec × 200 bytes = 20 KB/sec
Bandwidth saved: 90%

Memory Impact

Tag filtering has minimal memory overhead:

Tags stored temporarily during routing
No persistent tag index maintained
Memory scales with concurrent connection count, not message history

Standard Pusher Clients

Clients without filter support (pusher-js, Laravel Echo):

Can still subscribe to channels normally
Receive all messages (no filtering applied)
Cannot specify filter criteria
Tags are still included if enable_tags: true

To use filtering, use @sockudo/client or implement the filter protocol.

Filter Protocol Reference

{
  "event": "pusher:subscribe",
  "data": {
    "channel": "market-data",
    "auth": "...",
    "filter": {
      "type": "and",
      "filters": [
        {
          "type": "eq",
          "key": "symbol",
          "value": "BTC"
        },
        {
          "type": "gt",
          "key": "volume",
          "value": "1000"
        }
      ]
    }
  }
}

Filter Types

Type	Fields	Example
`eq`	`key`, `value`	`{"type":"eq","key":"symbol","value":"BTC"}`
`neq`	`key`, `value`	`{"type":"neq","key":"status","value":"inactive"}`
`gt`	`key`, `value`	`{"type":"gt","key":"price","value":"1000"}`
`gte`	`key`, `value`	`{"type":"gte","key":"volume","value":"100"}`
`lt`	`key`, `value`	`{"type":"lt","key":"quantity","value":"50"}`
`lte`	`key`, `value`	`{"type":"lte","key":"age","value":"30"}`
`in`	`key`, `values`	`{"type":"in","key":"symbol","values":["BTC","ETH"]}`
`nin`	`key`, `values`	`{"type":"nin","key":"region","values":["US","EU"]}`
`exists`	`key`	`{"type":"exists","key":"urgent"}`
`nexists`	`key`	`{"type":"nexists","key":"deprecated"}`
`prefix`	`key`, `value`	`{"type":"prefix","key":"symbol","value":"BTC"}`
`suffix`	`key`, `value`	`{"type":"suffix","key":"symbol","value":"USD"}`
`and`	`filters`	`{"type":"and","filters":[...]}`
`or`	`filters`	`{"type":"or","filters":[...]}`

Monitoring & Debugging

Server Metrics

Check filter effectiveness:

curl http://localhost:9601/metrics | grep filter

Key metrics:

sockudo_filtered_messages_total - Messages filtered
sockudo_filter_matches_total - Successful filter matches
sockudo_filter_evaluations_total - Total filter evaluations

Client Debugging

Enable debug output:

const pusher = new Pusher("app-key", {
  wsHost: "127.0.0.1",
  wsPort: 6001,
  forceTLS: false,
  enableLogging: true
});

const channel = pusher.subscribe(
  "market-data",
  Filter.eq("symbol", "BTC")
);

Console output:

[Sockudo] Subscribing to market-data with filter: {"type":"eq","key":"symbol","value":"BTC"}
[Sockudo] Subscription succeeded: market-data
[Sockudo] Message received on market-data: trade

Testing Filters

Test filter expressions before deploying:

// Test in development
const testFilter = Filter.and(
  Filter.eq("symbol", "BTC"),
  Filter.gt("volume", "1000")
);

console.log("Filter:", JSON.stringify(testFilter, null, 2));

// Subscribe and verify
const channel = pusher.subscribe("test-channel", testFilter);
channel.bind("test-event", (data) => {
  console.log("Received filtered message:", data);
});

Best Practices

✅ Do

Design tags for filtering: Plan tag structure based on filter needs
Use descriptive tag names: Clear, consistent naming (e.g., "user_id", not "uid")
Keep tag values simple: Strings only, avoid nested objects
Test filter combinations: Verify logic before production
Monitor bandwidth savings: Track effectiveness with metrics
Disable tag forwarding if clients don't need tags (enable_tags: false)

❌ Don't

Don't filter client-side when server-side filtering is available
Don't use unique values as tags: Defeats filtering purpose
Don't duplicate payload in tags: Keep tags separate from data
Don't over-complicate filters: Simple filters perform better
Don't forget to tag messages: Untagged messages go to all subscribers
Don't change tag schema without updating clients

Migration from Client-Side Filtering

Before (client-side filtering):

// All messages sent to client
const channel = pusher.subscribe("market-data");
channel.bind("trade", (data) => {
  // Filter on client
  if (data.symbol === "BTC") {
    console.log("BTC trade:", data);
  }
});
// Wastes bandwidth: Receives all messages, uses only some

After (server-side filtering):

// Only BTC messages sent to client
const channel = pusher.subscribe(
  "market-data",
  Filter.eq("symbol", "BTC")
);
channel.bind("trade", (data) => {
  console.log("BTC trade:", data);
});
// Efficient: Receives only relevant messages

Migration steps:

Enable tag filtering in server config
Update publishers to include tags
Update clients to use @sockudo/client with filters
Remove client-side filtering logic
Monitor bandwidth savings