FLYsafe.live - v1 - Cross Tenant Partnerships

Problem Statement

Two tenants want to cooperate so that the Home Tenant can see the Partner Tenant's live drone positions on their RTM frontend.

Core Principle

The Partner Tenant's drone telemetry is already in the shared Kafka cluster as {partner_tenant}.telemetry.enriched, produced by the Partner Tenant's FlightBinder (Flink) pipeline. It just needs to be routed to the Home Tenant's EMQX — already in canonical RTTP schema, no additional transform required at consume time.

Why Kafka — Not a Direct EMQX Bridge

A direct EMQX-to-EMQX bridge was considered first: Home Tenant's EMQX subscribes to Partner Tenant's EMQX and re-publishes messages locally. EMQX 5 supports this natively and it has lower latency.

Why it was rejected: The Home Tenant's EMQX needs a username and password on the Partner Tenant's EMQX to connect. This creates a cross-tenant credential dependency — the Partner Tenant must create a dedicated EMQX user for every partner, transfer those credentials to the Home Tenant, and delete the user on revoke. The Partner Tenant has an operational dependency on every active partnership.

With the Kafka approach, the Partner Tenant does nothing. Its telemetry is already in Kafka. The KafkaConnector is created and deleted entirely within shared-kafka by the Home Tenant's controller — the Partner Tenant is never touched.

Data Model

Each tenant's PostgreSQL database has a partnerships schema with two tables.

approved_partners — allowlist

Manually populated by a tenant admin. Only tenants listed here can submit a partnership request. Acts as an access control gate before any request is considered.

partner_tenant  TEXT PRIMARY KEY
added_at        TIMESTAMPTZ NOT NULL DEFAULT now()

approval_requests — request lifecycle log

Every incoming partnership request with full status history.

id                  SERIAL PRIMARY KEY
requesting_tenant   TEXT NOT NULL
status              TEXT NOT NULL DEFAULT 'pending'
                    CHECK (status IN ('pending', 'enabled', 'suspended', 'denied'))
requested_at        TIMESTAMPTZ NOT NULL DEFAULT now()
status_changed_at   TIMESTAMPTZ
notes               TEXT

Status state machine

pending ──approve──▶ enabled ──suspend──▶ suspended
                       ▲                      │
                       └────────resume─────────┘

pending ──deny──▶ denied  ← terminal, no further transitions

Component Diagram

┌───────────────────────── Home Tenant namespace ─────────────────────────────┐
│                                                                               │
│  RTM Frontend (Next.js)                                                       │
│      │  POST   /partnerships/approved-partners   → add to allowlist           │
│      │  POST   /partnerships/requests            → submit request (partner)   │
│      │  PATCH  /partnerships/requests/{id}/approve → admin approves           │
│      │  PATCH  /partnerships/requests/{id}/deny    → admin denies             │
│      ▼                                                                        │
│  Partnership Controller  (Python FastAPI)                NEW SERVICE      │
│      │  ServiceAccount: partnership-controller-svc                            │
│      │  ├─ reads:  emqx-user-svc-mqtt-client (k8s secret, own ns)            │
│      │  ├─ reads:  postgres-partnership-credentials (k8s secret, own ns)      │
│      │  ├─ writes: approved_partners, approval_requests (PostgreSQL, own ns)  │
│      │  └─ calls:  Kubernetes API → KafkaConnector in shared-kafka            │
│      │             (no outbound calls to other tenants)                       │
│                                                                               │
│  PostgreSQL                                                                   │
│      schema: partnerships                                                     │
│      tables: approved_partners, approval_requests                             │
│                                                                               │
│  Home Tenant EMQX  (tenant-emqx.{home_tenant}.svc:1883)                     │
│      topics (per active partner serial):                                      │
│        v1/partner/{partner_tenant}/aircraft/{serial}/drones/telemetry  ← position + status │
│        v1/partner/{partner_tenant}/aircraft/{serial}/drones/networkid  ← pilot + mission   │
│        v1/partner/{partner_tenant}/aircraft/{serial}/oi/declaration    ← OI boundary       │
│      ▼                                                                        │
│  RTM Frontend subscribes: v1/partner/+/aircraft/+/#                          │
│      renders partner drones: grey icon, "Partner: {partner_tenant}" label    │
│                                                                               │
└───────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────── shared-kafka namespace ──────────────────────────┐
│                                                                               │
│  {partner_tenant}.telemetry.enriched   (Kafka topic — already exists)        │
│  {partner_tenant}.networkid.dd         (Kafka topic — already exists)        │
│  {partner_tenant}.flightplan           (Kafka topic — already exists)        │
│      │                                                                        │
│      ▼                                                                        │
│  partner-{partner_tenant}-to-{home_tenant}-mqtt-sink  (KafkaConnector)  CREATED ON APPROVE  │
│      class:      Lenses MqttSinkConnector                                     │
│      reads:      {partner_tenant}.telemetry.enriched  ·  {partner_tenant}.networkid.dd  ·  {partner_tenant}.flightplan  │
│      transform:  InsertField  partner_tenant = "{partner_tenant}"             │
│      key:        Kafka message key = aircraft serial → builds per-serial path │
│      publishes:  v1/partner/{partner_tenant}/aircraft/{serial}/drones/telemetry  → Home Tenant EMQX  │
│                  v1/partner/{partner_tenant}/aircraft/{serial}/drones/networkid  → Home Tenant EMQX  │
│                  v1/partner/{partner_tenant}/aircraft/{serial}/oi/declaration    → Home Tenant EMQX  │
│                                                                               │
└───────────────────────────────────────────────────────────────────────────────┘

┌──────────────────────── Partner Tenant namespace ───────────────────────────┐
│                                                                               │
│  (Partner Tenant's Flink, EMQX, and primary pipeline are never touched)      │
│                                                                               │
│  Partner Tenant's Partnership Controller                                      │
│      POST /partnerships/requests  → Partner Tenant submits a request to      │
│      Home Tenant's controller  (Partner Tenant calls Home Tenant's API)      │
│                                                                               │
└───────────────────────────────────────────────────────────────────────────────┘

Topic Mappings

Topic patterns are configured in Helm values — no image rebuild required to add a topic. {partner_tenant} is substituted at connector-creation time. The Kafka message key (aircraft serial) is used by the connector to construct the per-aircraft MQTT path — one connector handles all aircraft for a partnership via a multi-statement KCQL.

connector:
  topicMappings:
    - kafkaTopicTemplate: "{partner_tenant}.telemetry.enriched"
      mqttTopicTemplate: "v1/partner/{partner_tenant}/aircraft/{serial}/drones/telemetry"

    - kafkaTopicTemplate: "{partner_tenant}.networkid.dd"
      mqttTopicTemplate: "v1/partner/{partner_tenant}/aircraft/{serial}/drones/networkid"

    - kafkaTopicTemplate: "{partner_tenant}.flightplan"
      mqttTopicTemplate: "v1/partner/{partner_tenant}/aircraft/{serial}/oi/declaration"

MQTT Topic Design

Partner topics use a partner root that is entirely separate from own-drone topics. Because every EMQX broker is already scoped to a single tenant, the {tenant} segment is implicit in the connection endpoint — repeating it in the partner path would be redundant. Instead, the literal keyword partner occupies segment 2 in place of {tenant}, giving the frontend an immediate, unambiguous signal that a message originates from a cross-tenant feed without inspecting the payload.

RBAC Model

The Partnership Controller's ServiceAccount lives in the tenant namespace but manages KafkaConnectors in shared-kafka. A cross-namespace Role and RoleBinding are created in shared-kafka:

Home Tenant namespace                     shared-kafka namespace
─────────────────────────────             ──────────────────────────────────────
ServiceAccount                            Role
  partnership-controller-svc  ←────────   partnership-controller-{home_tenant}
                                            rules:
                                              kafkaconnectors:
                                                get, list, create,
                                                update, patch, delete

                                          RoleBinding
                                            subject: SA from Home Tenant ns
                                            roleRef: partnership-controller-{home_tenant}

Authentication

All endpoints on the partnership-controller require a valid Keycloak JWT. Token validation happens in the FastAPI application — no ingress-level auth is involved, making it consistent whether the controller is reached via its public ingress or over ClusterIP.

Keycloak Realm and Group Structure

A single Keycloak realm (flysafe at auth2.airmarket.io) serves all tenants. Tenants are separated by Keycloak Groups with an admins subgroup per tenant:

/airmarket
    /admins    ← full access to all partnership endpoints
    /users     ← no access to the partnership API
/sandbox
    /admins    ← full access to all partnership endpoints
    /users     ← no access to the partnership API

Keycloak embeds group membership in the JWT groups claim:

{
  "iss": "https://auth2.airmarket.io/realms/flysafe",
  "groups": ["/sandbox", "/sandbox/admins"],
  "sub": "user-uuid"
}

Endpoint Access Control

JWT Validation Flow

• 1
  Extracts the Bearer token from the Authorization header.
• 2
  Fetches Keycloak's JWKS from KEYCLOAK_JWKS_URL (cached 5 minutes in memory).
• 3
  Matches the token's kid header to a key in the JWKS.
• 4
  Verifies RS256 signature, expiry, and iss against KEYCLOAK_ISSUER.
• 5
  Checks the groups claim for the required group membership. Returns 403 if absent.
• 6
  On key-rotation miss: forces an immediate JWKS refresh and retries once before returning 401.

Both KEYCLOAK_JWKS_URL and KEYCLOAK_ISSUER are injected via Helm values — no image rebuild is required to change the Keycloak realm or base URL.

Flow 1 — Admin Populates the Allowlist

Done once per trusted partner. Only tenants on this list can submit requests — unknown tenants are rejected immediately with 403.

• 1
  Home Tenant's admin calls POST /partnerships/approved-partners with { "partner_tenant": "{partner_tenant}" }.
• 2
  Controller writes INSERT INTO approved_partners (partner_tenant='{partner_tenant}') → 201 OK.

Flow 2 — Partner Tenant Submits a Partnership Request

• 1
  Partner Tenant's admin calls POST https://partnerships.{home_tenant}.stage.flysafe.live/partnerships/requests with { "requesting_tenant": "{partner_tenant}" }.
• 2
  Controller checks approved_partners for {partner_tenant}. Not found → 403 (stops here). Found → continue.
• 3
  Controller checks approval_requests for an existing pending or suspended request from {partner_tenant}.
  pending exists → 409 (re-submit allowed only after denied).
  suspended exists → 409 with message to use /resume on the existing request.
  Neither exists → continue.
• 4
  INSERT INTO approval_requests (requesting_tenant='{partner_tenant}', status='pending') → 201 { "id": 7, "status": "pending", ... }

Flow 3 — Home Tenant Admin Manages the Partnership

• 1
  Home Tenant's admin calls GET /partnerships/requests?status=pending to review the queue.
• 2
  Approve: PATCH /partnerships/requests/7/approve
  Controller verifies request is pending.
  Builds and creates KafkaConnector in shared-kafka (PATCHes if 409).
  Updates status → enabled.
  Returns 200 { "status": "enabled", "partner_tenant": "{partner_tenant}", "mqtt_wildcard": "v1/partner/{partner_tenant}/aircraft/+/#" }.
• 3
  Deny: PATCH /partnerships/requests/7/deny
  Controller verifies request is pending. Denied requests are terminal — no further transitions allowed.
  Deletes KafkaConnector if present.
  Updates status → denied.
  Returns 200 { "status": "denied" }. The Partner Tenant must submit a new request to try again.
• 4
  Suspend: PATCH /partnerships/requests/7/suspend
  Controller verifies request is enabled.
  Deletes KafkaConnector — streaming stops immediately. Tenant stays on allowlist.
  Updates status → suspended.
  Returns 200 { "status": "suspended" }.
• 5
  Resume: PATCH /partnerships/requests/7/resume
  Controller verifies request is suspended.
  Recreates KafkaConnector — streaming restarts within seconds.
  Updates status → enabled.
  Returns 200 { "status": "enabled", "partner_tenant": "{partner_tenant}", "mqtt_wildcard": "v1/partner/{partner_tenant}/aircraft/+/#" }.

What Is Not Touched During Any Status Change

REST API Surface

All endpoints at https://partnerships.{tenant}.{env}.flysafe.live/

Allowlist management

Partnership requests

End-to-End Data Flow (Partnership Active)

Partner Tenant's Drone
    │
    ▼  (existing, unchanged)
Partner Tenant's DJI Integration Service
    │  MQTT publish
    ▼
Partner Tenant EMQX  →  Partner Tenant's FlightBinder (Flink)  +  Partner Tenant's OI Conformance Tracker
    │
    ▼
Kafka (shared-kafka namespace):
    {partner_tenant}.telemetry.enriched   ← canonical RTTP telemetry, already decoded
    {partner_tenant}.networkid.dd         ← deduplicated networkid (pilot, mission, waypoints)
    {partner_tenant}.flightplan           ← OI declaration + conformance state
    │
    │  ← partnership active: KafkaConnector exists
    ▼
partner-{partner_tenant}-to-{home_tenant}-mqtt-sink  (KafkaConnector, shared-kafka)  ← CREATED ON APPROVE
    │  transform: InsertField  partner_tenant = "{partner_tenant}"
    │  key: Kafka message key = aircraft serial → builds per-aircraft MQTT path
    │  connects to: tenant-emqx.{home_tenant}.svc.cluster.local:1883
    ▼
Home Tenant EMQX
    │  v1/partner/{partner_tenant}/aircraft/{serial}/drones/telemetry   10 Hz
    │  v1/partner/{partner_tenant}/aircraft/{serial}/drones/networkid   ~1 Hz
    │  v1/partner/{partner_tenant}/aircraft/{serial}/oi/declaration     on change
    ▼
RTM Frontend (Home Tenant)  via WebSocket  wss://emqx-ws.{home_tenant}.stage.flysafe.live/mqtt
    │  wildcard sub: v1/partner/+/aircraft/+/#
    ▼
Map renders:
    ● Blue icon  — Home Tenant's own drones   (v1/{home_tenant}/aircraft/+/drones/telemetry)
    ● Grey icon  — Partner Tenant's drones    (v1/partner/{partner_tenant}/aircraft/{serial}/…)
      label: "Partner: {partner_tenant}"  ·  pilot + mission from networkid
      OI boundary overlay from oi/declaration + conformance colour