Cross-Tenant Drone Visibility

Problem Statement

Two tenants (A and B) want to cooperate so that Tenant A can see Tenant B's live drone positions on their RTM frontend.

Core Principle

B's drone telemetry is already in the shared Kafka cluster as B.telemetry.enriched, produced by B's FlightBinder (Flink) pipeline. It just needs to be routed to A'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: Tenant A's EMQX subscribes to Tenant B's EMQX and re-publishes messages locally. EMQX 5 supports this natively and it has lower latency.

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

With the Kafka approach, Tenant B does nothing. Its telemetry is already in Kafka. The KafkaConnector is created and deleted entirely within shared-kafka by Tenant A's controller — Tenant B 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

┌─────────────────────────── Tenant A namespace ──────────────────────────────┐
│                                                                               │
│  RTM Frontend (Next.js)                                                       │
│      │  POST   /partnerships/approved-partners   → add to allowlist           │
│      │  POST   /partnerships/requests            → submit request (by B)      │
│      │  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                             │
│                                                                               │
│  Tenant-A EMQX  (tenant-emqx.A.svc:1883)                                    │
│      topics (per active partner serial):                                      │
│        v1/partner/B/aircraft/{serial}/drones/telemetry  ← position + status │
│        v1/partner/B/aircraft/{serial}/drones/networkid  ← pilot + mission   │
│        v1/partner/B/aircraft/{serial}/oi/declaration    ← OI boundary       │
│      ▼                                                                        │
│  RTM Frontend subscribes: v1/partner/+/aircraft/+/#                          │
│      renders B's drones: grey icon, "Partner: B" label                       │
│                                                                               │
└───────────────────────────────────────────────────────────────────────────────┘

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

┌─────────────────────────── Tenant B namespace ──────────────────────────────┐
│                                                                               │
│  (B's Flink, EMQX, and primary pipeline are never touched)                   │
│                                                                               │
│  B's Partnership Controller                                                   │
│      POST /partnerships/requests  → B submits a request to A's controller    │
│      (B calls A's API — not the other way around)                            │
│                                                                               │
└───────────────────────────────────────────────────────────────────────────────┘

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:

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

                                          RoleBinding
                                            subject: SA from Tenant-A ns
                                            roleRef: partnership-controller-A

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
  A's admin calls POST /partnerships/approved-partners with { "partner_tenant": "B" }.
• 2
  Controller writes INSERT INTO approved_partners (partner_tenant='B') → 201 OK.

Flow 2 — B Submits a Partnership Request

• 1
  B's admin calls POST https://partnerships.A.stage.flysafe.live/partnerships/requests with { "requesting_tenant": "B" }.
• 2
  Controller checks approved_partners for "B". Not found → 403 (stops here). Found → continue.
• 3
  Controller checks approval_requests for an existing pending or suspended request from "B".
  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='B', status='pending') → 201 { "id": 7, "status": "pending", ... }

Flow 3 — A's Admin Manages the Partnership

• 1
  A'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": "B", "mqtt_wildcard": "v1/partner/B/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" }. B 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": "B", "mqtt_wildcard": "v1/partner/B/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)

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