Digital Twin Service

Implementation reference for the digital-twin. For its architectural role, see Digital Twin Service Architecture; for the domain model, see Building Management.

A Rust / Axum service. It is the spatial source of truth: it stores buildings and rooms, and on every change it synchronises a lightweight clone to the dependent services — asynchronously over Kafka for registration, synchronously over REST for later edits (see Downstream Synchronisation below). The room set is write-once: the upload creates it and nothing edits it afterwards, so the twin cannot drift from the model it was built from.

Source Layout

For the conceptual view — which module is core, which is a port, which is an adapter, and why — see Digital Twin Service Architecture. What follows is the wiring as it actually exists in the crate: concrete types, the layer each one sits in, and what calls what at run time.

graph TB
    REQ(["HTTP request"])
    TICK(["Tokio tick"])
    KAFKAIN(["Kafka message"])

    subgraph l1["LAYER 1 — driving adapters"]
        ROUTER["lib.rs\nbuild_router assembles the Router"]
        MW["adapters/metrics.rs · ratelimit.rs\nRouter::layer, cross-cutting"]
        EXTRACT["adapters/driving/http_api/claims.rs\nFromRequestParts for GatewayClaims"]
        HANDLERS["adapters/driving/http_api/controllers.rs\n8 handlers"]
        ERRMAP["adapters/driving/http_api/exceptions.rs\nIntoResponse for DomainError"]
        WORKER["adapters/driving/worker.rs\nspawn: loop over provision_next"]
        KCONS["adapters/driving/kafka_consumer.rs\nspawn: consume, call resolve"]
    end

    subgraph l2["LAYER 2 — core, use cases (service/)"]
        UCB["Buildings\nget · list_for_domain · counts · domains_of\nupdate"]
        UCP["Provisioning\naccept · provision_next · resolve · status"]
        AUTHZ["authz.rs\nis_member_of · can_edit_domains\nscope_to_memberships"]
    end

    subgraph l2p["LAYER 2 — core, ports (service/ports.rs)"]
        PS["trait BuildingStore"]
        PQ["trait UploadQueue"]
        PD["trait DownstreamSync"]
        PE["trait RegistrationEvents"]
    end

    subgraph l1e["LAYER 1 — core, entities (domain/)"]
        ENT["Building · Room · Coordinates · Dimensions\nGatewayClaims · AcceptedUpload · UploadStatus\nDomainError"]
    end

    subgraph l3["LAYER 3 — driven adapters (adapters/driven/)"]
        IDB["persistence/db.rs\nMongoBuildings"]
        IJOB["persistence/jobs.rs\nMongoUploadQueue"]
        IOUT["outbound.rs\nOutboundConfig"]
        IKPROD["kafka_producer.rs\nKafkaEventProducer"]
    end

    subgraph l4["LAYER 4 — frameworks and drivers"]
        MONGO[("twin-db\nbuildings · pending_uploads")]
        SENSOR["telemetry"]
        CONTRACTS["dashboard"]
        KAFKA[("Kafka\nbuilding-registration-requested/-completed")]
        POLICY["backend/libs/auth-policy\npolicy.cedar + schema"]
    end

    REQ --> ROUTER --> MW --> EXTRACT --> HANDLERS
    HANDLERS --> ERRMAP
    TICK --> WORKER
    KAFKAIN --> KCONS

    HANDLERS --> UCB
    HANDLERS --> UCP
    WORKER --> UCP
    KCONS --> UCP

    UCB --> AUTHZ
    UCB --> PS
    UCB --> PD
    UCP --> PS
    UCP --> PQ
    UCP --> PD
    UCP --> PE

    UCB --> ENT
    UCP --> ENT
    HANDLERS --> ENT

    IDB -.->|impl| PS
    IJOB -.->|impl| PQ
    IOUT -.->|impl| PD
    IKPROD -.->|impl| PE

    IDB --> MONGO
    IJOB --> MONGO
    IOUT --> SENSOR
    IOUT --> CONTRACTS
    IKPROD --> KAFKA
    KAFKA --> SENSOR
    SENSOR --> KAFKA
    AUTHZ -.->|"include_str!, compile time"| POLICY

Two things to read off it. First, adapters/driving/http_api, adapters/driving/worker.rs, and adapters/driving/kafka_consumer.rs all stop at layer 2 — none has an edge into layer 3, which is the grep the crate enforces. Second, the only solid arrows crossing from layer 3 into layer 2 are the dashed impl ones, pointing at traits: adapters/driven depends on service, never the reverse. Note that Provisioning is now entered from three directions (HANDLERS, WORKER, KCONS) rather than two — the same use case, three ways in.

Configuration

Data Model

struct Building {
    id: String,        // server-generated UUID (uuid::Uuid::new_v4); unique
    name: String,
    rooms: Vec<Room>,
    domains: Vec<String>,
}

struct Room {
    id: String,         // client-supplied on /register
    name: String,
    capacity: f64,
    position: Coordinates,     // { x, y, z }
    dimensions: Dimensions,    // { width, height, depth }
    color: Option<String>,
}

• The building id is generated server-side on POST /register; a client-supplied one is ignored. Room ids are client-supplied and come from the uploaded description — they are the only ids a client chooses, and there is no later route to change them.
• The Mongo collection carries a unique index on id and a non-unique index on domains (Building.find({ domains: X }) stays an index seek).
• The pending_uploads queue carries a unique index on id (status polls and resolves) and one on status + leased_until (the worker’s claim).
• Finished uploads get a finished_at, and a TTL index removes them an hour later. The record must outlive the browser’s status poll (up to 30 s), or a successful upload times out; deleting on completion fails every such poll. MongoDB’s TTL sweep runs every 60 s, so removal lags by up to a minute.
• On startup the queue backfills finished_at onto uploads finished before expiry existed; a TTL index never removes a record without its field.
• No internal database fields (_id, a document version, timestamps) leak into API responses — the Building/Room structs only carry fields a client actually reads.

HTTP API

Name Normalisation

Free functions in src/domain/building.rs.

• normalize_building_name(name, id) resolves to the trimmed name, else id, else the literal "Building".
• normalize_room_name(name, id) trims the room name, falling back to the room id.
• Every building load (GET, and every mutating handler’s initial fetch) backfills blank names and persists the fix if anything changed — so a document with a blank name gets healed the next time it’s touched, without a separate migration.

Downstream Synchronisation

A building’s structure is mirrored to telemetry and dashboard, but not the same way: registration announces over Kafka and waits on telemetry’s own outcome; a later building update (rename, re-domain, max-temperature) still calls telemetry directly over REST, synchronously, in-request. Contracts-service preference seeding is REST and best-effort in both cases.

sequenceDiagram
    autonumber
    participant C as Client
    participant T as digital-twin (API)
    participant Q as pending_uploads
    participant W as digital-twin (worker)
    participant DB as twin-db
    participant K as Kafka
    participant S as telemetry
    participant KC as digital-twin (kafka_consumer)
    participant Ctr as dashboard
    C->>T: POST /register { name, rooms, domains }
    Note over T: normalise names, validate geometry, assign UUID
    T->>Q: enqueue PendingUpload (durable)
    T-->>C: 202 { buildingId }
    W->>Q: claim (lease, atomic)
    W->>DB: upsert
    W->>K: publish building-registration-requested
    Note over W,K: publish returns immediately -- does not wait on telemetry
    W->>Ctr: POST /preferences/init/:id
    Note over W,Ctr: best-effort — errors logged, swallowed
    K->>S: building-registration-requested
    S->>S: write threshold clone (upsert)
    S->>K: publish building-registration-completed { status: "ready" }
    K->>KC: building-registration-completed
    KC->>Q: resolve -- mark ready
    C->>T: GET /building/:id/status
    T-->>C: { status: "ready" }

• Registration announcement (KafkaEventProducer::publish_requested, src/adapters/driven/kafka_producer.rs): publishes building-registration-requested keyed by buildingId, carrying { buildingId, name, rooms: [{ id, name }] }. A publish failure fails the provisioning job the same way a REST failure used to; a successful publish does not — the upload stays pending until telemetry’s own completion event resolves it.
• Registration resolution (adapters/driving/kafka_consumer.rs): consumes building-registration-completed, calling resolve(buildingId, error), which is what actually marks the upload ready or failed. Idempotent under redelivery — resolving the same id twice sets the same status twice.
• Sensor threshold clone on a building update (sync_building_clone, src/adapters/driven/outbound.rs): a PUT carrying { name, rooms: [{ id, name }], maxTemperature? }. Skipped entirely under NODE_ENV=test, and returns an error on a non-2xx response — which fails the mutating request itself, since the route is already synchronous. Forwards the caller’s x-gateway-claims header verbatim — telemetry reads the same mesh-injected identity twin trusts; mTLS already authenticates the hop, the header just carries who the original caller was.
• Contracts preference seeding (init_building_preferences): a POST, best-effort — logged and swallowed on any error, never blocks building creation.
• Provisioning-failure cleanup and alert (Provisioning::fail, private, called by both provision_next and resolve): marks the job failed, calls notification’s POST /trigger (notify_provisioning_failed, src/adapters/driven/outbound.rs) as a fixed system caller — same pattern as notification’s own event-listener path, see Notification Service Architecture — then deletes the twin (BuildingStore::delete). Both steps are skipped unless mark_failed actually claimed a still-pending upload — a failure reported for an upload that was already resolved is logged and dropped, so a re-announced building (POST /building/:id/sync) that a consumer rejects can never delete a live twin. Notify happens before delete: /trigger resolves the building’s domains by calling back into digital-twin’s /domain/:building, so the twin must still exist when that call lands. The HTTP call is best-effort — logged and swallowed on error, never blocks the failure resolution itself.

Error Handling

DomainError (src/domain/error.rs) is mapped to HTTP by src/adapters/driving/http_api/exceptions.rs’s IntoResponse impl, mapping to a stable { type, message } JSON shape and status code: Validation → 400, NotFound → 404, Unauthorized → 401, Forbidden → 403. Unexpected errors (a Mongo failure, a Cedar entity-build failure) are logged via log::error! and mapped to a generic 500 — every error is logged, and no internal detail leaks past the generic 500 message.

Metrics

track_metrics is an Axum Router::layer, which wraps the entire request/response cycle regardless of how the inner handler responds — MatchedPath (the route template, e.g. /building/{id}, not the literal path with real ids) is available to layers added this way, so every registered route is observed regardless of the response it produces. Verified live: registering a building and then curling /metrics shows real http_requests_total/http_request_duration_seconds/http_error_requests_total series with the correct method/route/status_code labels.

Cedar Authorization

src/service/authz.rs embeds the shared backend/libs/auth-policy bundle via include_str! — resolved at compile time, the same way Go’s //go:embed does. Only Python’s binding reads schema.cedarschema/policy.cedar from disk, at process start, since it has no equivalent embed mechanism. Compile-time embedding here means the production Docker image doesn’t need backend/libs/auth-policy present at all; only the builder stage does, since the bytes are already baked into the compiled binary by the time cargo build finishes. See Auth Policy for the full Cedar model (pre-expanded role-tier sets, .contains() vs in, the golden conformance suite).

digital-twin uses two of Cedar’s five shared actions: Read (is_member_of/scope_to_memberships) and Edit (can_edit_domains, tried against every one of a building’s domains — permits if any qualifies). A third function, authorize_any, exists purely so the golden conformance suite can exercise all five actions through this binding too, matching the precedent set by agent’s can_override_model.