User Durable Object - Per-Property Data Segregation

The UserDurableObject provides isolated SQLite storage for each property, ensuring complete data separation between tenants.

Key Files

• crates/bf_user/src/durable_object_sqlx.rs - DO implementation
• crates/bf_user/src/proxy.rs - Proxy for calling DOs from workers
• crates/bf_user/src/rpc.rs - RPC request/response types
• crates/bf_user/migrations/ - DO database schema

Architecture Overview

┌─────────────────────────────────────────────────────────────────────────┐
│                         bf_resident Worker                              │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   Request: POST /user/properties/{property_id}/guests                   │
│                                                                         │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │                    bf_user::proxy()                             │  │
│   │                                                                  │  │
│   │   1. Get DO namespace: env.durable_object("USER_DO")            │  │
│   │   2. Get DO ID: namespace.id_from_name(property_id)             │  │
│   │   3. Get DO stub: id.get_stub()                                 │  │
│   │   4. Create RPC request                                          │  │
│   │   5. POST to DO: stub.fetch("/rpc", request)                    │  │
│   │   6. Parse RpcResponse                                           │  │
│   │                                                                  │  │
│   └─────────────────────────────────────────────────────────────────┘  │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    │ RPC over HTTP
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│              UserDurableObject (property_id = "prop_123")               │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │                     Isolated SQLite DB                          │  │
│   │                                                                  │  │
│   │   • property_metadata (1 row)                                   │  │
│   │   • tenants (cached from D1)                                    │  │
│   │   • guests                                                       │  │
│   │   • pin_codes                                                    │  │
│   │   • access_logs                                                  │  │
│   │   • energy_readings                                              │  │
│   │   • device_states                                                │  │
│   │   • notifications                                                │  │
│   │   • device_tokens                                                │  │
│   │   • credentials                                                  │  │
│   │                                                                  │  │
│   └─────────────────────────────────────────────────────────────────┘  │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

DO Implementation

Structure

// crates/bf_user/src/durable_object_sqlx.rs

#[durable_object]
pub struct UserDurableObject {
    state: Arc<State>,
    env: Env,
    conn: DOConnection,
    migrations_run: Cell<bool>,
}

#[durable_object]
impl DurableObject for UserDurableObject {
    fn new(state: State, env: Env) -> Self {
        let conn = DOConnection::new(&state).expect("Failed to create DO connection");
        
        Self {
            state: Arc::new(state),
            env,
            conn,
            migrations_run: Cell::new(false),
        }
    }

    async fn fetch(&self, req: Request) -> worker::Result<Response> {
        self.ensure_initialized().await?;
        self.route_request(req).await
    }
    
    async fn alarm(&self) -> worker::Result<Response> {
        // Handle scheduled tasks (token refresh)
        self.handle_alarm().await
    }
}

Initialization

Migrations run automatically on first request:

async fn ensure_initialized(&self) -> worker::Result<()> {
    if !self.migrations_run.get() {
        // Run SQL migrations
        run_migrations(&mut self.conn.clone(), migrations!("./migrations")).await?;

        // Clean up old access logs (retention policy)
        if let Ok(deleted) = storage::cleanup_old_logs(&self.conn, LOG_RETENTION_SECS).await {
            if deleted > 0 {
                console_log!("Cleaned up {} old access log entries", deleted);
            }
        }

        // Schedule token refresh alarm if needed
        self.ensure_alarm_scheduled().await;

        self.migrations_run.set(true);
    }
    Ok(())
}

Request Routing

async fn route_request(&self, mut req: Request) -> worker::Result<Response> {
    let path = req.path();

    match (req.method(), path.as_str()) {
        (Method::Post, "/rpc") => self.handle_rpc(&mut req).await,
        (Method::Get, "/health") => Response::ok("OK"),
        _ => Response::error("Not Found", 404),
    }
}

Database Schema

property_metadata

Single row caching property info from D1:

CREATE TABLE property_metadata (
    property_id TEXT PRIMARY KEY,
    portfolio_id TEXT,
    address TEXT,
    unit TEXT,
    bedrooms INTEGER,
    status TEXT,           -- PropertyStatus enum
    last_synced INTEGER    -- Unix timestamp
);

tenants

Cached tenant list:

CREATE TABLE tenants (
    email TEXT PRIMARY KEY,
    is_primary INTEGER,    -- 0 or 1
    role TEXT,             -- 'Tenant' or 'PrimaryTenant'
    access_granted INTEGER,
    added_at INTEGER
);

guests

Per-property guest access:

CREATE TABLE guests (
    id TEXT PRIMARY KEY,
    email TEXT UNIQUE,
    name TEXT NOT NULL,
    phone_number TEXT,
    access_start INTEGER,
    access_end INTEGER,
    pin_code TEXT,
    is_permanent INTEGER,
    invited_by TEXT,
    invited_at INTEGER,
    status TEXT,           -- JSON: GuestStatus
    permissions TEXT,      -- JSON: GuestPermissions
    created_at INTEGER
);

pin_codes

Yale lock PIN codes (supports multi-lock via lock_id):

CREATE TABLE pin_codes (
    id TEXT PRIMARY KEY,
    code TEXT NOT NULL,
    name TEXT,
    user_email TEXT,
    access_type TEXT,      -- 'always', 'temporary', 'recurring'
    valid_from INTEGER,
    valid_until INTEGER,
    valid_time_start TEXT, -- 'HH:MM'
    valid_time_end TEXT,
    valid_days TEXT,       -- 'MO,TU,WE,TH,FR'
    status TEXT,           -- PinCodeStatus enum
    yale_state TEXT,       -- Yale sync state
    created_at INTEGER,
    created_by TEXT,
    lock_id TEXT           -- Target lock device ID (NULL = "Front Door Lock")
);

Multi-Lock PIN Code Targeting

When creating a PIN code, the lock_id field determines which lock receives the code. The resolve_target_lock helper resolves the target:

1. If lock_id is provided, find that specific lock device
2. If lock_id is None, find a device named "Front Door Lock"
3. Fall back to the first available SmartLock device

access_logs

Lock/unlock history:

CREATE TABLE access_logs (
    id TEXT PRIMARY KEY,
    device_id TEXT NOT NULL,
    device_name TEXT,
    event_type TEXT NOT NULL, -- 'lock', 'unlock', 'pin_used', etc.
    method TEXT,              -- 'manual', 'pin', 'auto', 'remote'
    user_identifier TEXT,     -- Email or PIN code name
    timestamp INTEGER NOT NULL,
    metadata TEXT             -- JSON additional data
);

device_states

Current state of each device:

CREATE TABLE device_states (
    device_id TEXT PRIMARY KEY,
    device_type TEXT NOT NULL,
    state TEXT NOT NULL,      -- JSON: DeviceStateData
    last_updated INTEGER
);

notifications

User notifications:

CREATE TABLE notifications (
    id TEXT PRIMARY KEY,
    type TEXT NOT NULL,       -- NotificationType enum
    title TEXT NOT NULL,
    body TEXT,
    device_id TEXT,
    device_name TEXT,
    action_url TEXT,
    read INTEGER DEFAULT 0,
    created_at INTEGER
);

device_tokens

FCM push notification tokens:

CREATE TABLE device_tokens (
    id TEXT PRIMARY KEY,
    user_email TEXT NOT NULL,
    token TEXT UNIQUE NOT NULL,
    platform TEXT NOT NULL,   -- 'ios' or 'android'
    device_id TEXT,
    created_at INTEGER,
    last_used_at INTEGER
);

RPC Protocol

Request Types

Over 50 RPC operations defined in crates/bf_user/src/rpc.rs:

#[derive(Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RpcRequest {
    // Guest Management
    CreateGuest(CreateGuestRequest),
    UpdateGuest(UpdateGuestRequest),
    ListGuests,
    RevokeGuest { id: String },
    GetGuest { id: String },

    // PIN Code Management
    CreatePinCode(CreatePinCodeRequest),
    ListPinCodes,
    RevokePinCode { id: String },
    ActivatePinCode { id: String },
    ConfirmPinCodeDeletion { id: String },
    SyncPinCodes,

    // Access Logs
    ListAccessLogs(AccessLogQuery),
    CreateAccessLog(CreateAccessLogRequest),

    // Device States
    GetDeviceState { device_id: String },
    GetAllDeviceStates,
    RefreshDeviceStates,
    RefreshDeviceState { device_id: String },

    // Smart Device Control
    ExecuteSmartCapability {
        device_id: String,
        capability: DeviceCapability,
    },

    // Notifications
    ListNotifications(NotificationQuery),
    MarkNotificationRead { id: String },
    CreateNotification(CreateNotificationRequest),

    // Device Tokens
    RegisterDeviceToken(RegisterDeviceTokenRequest),
    UnregisterDeviceToken(UnregisterDeviceTokenRequest),

    // Energy
    GetEnergyReadings(EnergyReadingQuery),
    CreateEnergyReading(CreateEnergyReadingRequest),

    // Admin Operations (service-only)
    Seed(SeedRequest),
    MoveIn(MoveInRequest),
    MoveOut(MoveOutRequest),
    Backup(BackupRequest),

    // Sync Operations (service-only)
    SyncProperty(SyncPropertyRequest),
    SyncDevices(SyncDevicesRequest),
    SyncTenants(SyncTenantsRequest),
}

Response Types

#[derive(Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RpcResponse {
    Guest(Guest),
    Guests(Vec<Guest>),
    PinCode(PinCode),
    PinCodes(Vec<PinCode>),
    AccessLogs(Vec<AccessLog>),
    DeviceState(DeviceState),
    DeviceStates(Vec<DeviceState>),
    Notifications(Vec<Notification>),
    EnergyReadings(EnergyReadingsResponse),
    MoveInResponse(MoveInResponse),
    MoveOutResponse(MoveOutResponse),
    BackupResponse(BackupResponse),
    Success { message: String },
    Error { code: String, message: String },
}

Proxy Pattern

Basic Proxy Call

// crates/bf_user/src/proxy.rs

pub async fn proxy<R: DeserializeOwned>(
    env: &Env,
    property_id: &str,
    auth: Option<&ClerkAuth>,
    request: RpcRequest,
) -> bf_types::Result<R> {
    // 1. Get DO namespace
    let namespace = env.durable_object("USER_DO")?;

    // 2. Get DO ID from property_id
    let id = namespace.id_from_name(property_id)?;

    // 3. Get DO stub
    let stub = id.get_stub()?;

    // 4. Build request
    let body = serde_json::to_string(&request)?;
    let headers = worker::Headers::new();
    headers.set("X-Property-ID", property_id)?;
    headers.set("Content-Type", "application/json")?;
    
    if let Some(auth) = auth {
        headers.set("X-Clerk-Auth", &serde_json::to_string(auth)?)?;
    }

    let mut request_init = RequestInit::new();
    request_init.with_method(Method::Post);
    request_init.with_headers(headers);
    request_init.with_body(Some(body.into()));

    let do_request = Request::new_with_init(
        &format!("https://do/rpc"),
        &request_init
    )?;

    // 5. Call the DO
    let mut response = stub.fetch_with_request(do_request).await?;

    // 6. Parse response
    let rpc_response: RpcResponse = response.json().await?;
    
    match rpc_response {
        RpcResponse::Error { code, message } => {
            Err(AppError::from_rpc_error(&code, &message))
        }
        other => {
            let value = serde_json::to_value(other)?;
            serde_json::from_value(value).map_err(|e| AppError::Internal(e.to_string()))
        }
    }
}

Typed Proxy Functions

// Convenience functions for common operations

pub async fn list_guests(env: &Env, property_id: &str) -> bf_types::Result<Vec<Guest>> {
    proxy(env, property_id, None, RpcRequest::ListGuests).await
}

pub async fn create_guest(
    env: &Env,
    property_id: &str,
    request: CreateGuestRequest,
) -> bf_types::Result<Guest> {
    proxy(env, property_id, None, RpcRequest::CreateGuest(request)).await
}

pub async fn execute_capability(
    env: &Env,
    property_id: &str,
    auth: &ClerkAuth,
    device_id: &str,
    capability: DeviceCapability,
) -> bf_types::Result<DeviceState> {
    proxy(
        env,
        property_id,
        Some(auth),
        RpcRequest::ExecuteSmartCapability {
            device_id: device_id.to_string(),
            capability,
        },
    ).await
}

pub async fn move_in(
    env: &Env,
    property_id: &str,
    request: MoveInRequest,
) -> bf_types::Result<MoveInResponse> {
    proxy(env, property_id, None, RpcRequest::MoveIn(request)).await
}

Permission Checking

Property Status-Based Access

// crates/bf_user/src/permissions.rs

pub async fn check_permission(
    conn: &mut DOConnection,
    auth: Option<&ClerkAuth>,
    request: &RpcRequest,
    property_status: PropertyStatus,
) -> bf_types::Result<()> {
    // Service-only operations bypass auth
    if is_service_only_operation(request) {
        if auth.is_some() {
            return Err(AppError::Forbidden("Service-only operation".into()));
        }
        return Ok(());
    }

    let auth = auth.ok_or_else(|| AppError::Unauthorized("Authentication required".into()))?;
    let roles = get_user_roles(conn, auth).await?;

    // Check based on property status
    if property_status.is_vacant() {
        // Vacant: Admin, Installer can access
        if roles.contains(&UserRole::Admin) {
            return check_admin_permission(property_status, request);
        }
        if roles.contains(&UserRole::Installer) {
            return check_installer_permission(property_status, request);
        }
    } else {
        // Occupied: Tenant, Guest can access
        if roles.contains(&UserRole::Tenant) {
            return check_tenant_permission(property_status, request);
        }
        if roles.contains(&UserRole::Guest) {
            return check_guest_permission(conn, auth, property_status, request).await;
        }
    }

    Err(AppError::Forbidden("No permission for this operation".into()))
}

Guest Permission Model

pub struct GuestPermissions {
    pub monitor: bool,              // View device states
    pub front_door_and_camera: bool, // Lock/unlock, doorbell
    pub thermostat_control: bool,   // Temperature control
    pub security_control: bool,     // Alarm hub
    pub view_energy_usage: bool,    // Energy readings
}

async fn check_guest_permission(
    conn: &mut DOConnection,
    auth: &ClerkAuth,
    property_status: PropertyStatus,
    request: &RpcRequest,
) -> bf_types::Result<()> {
    let guest = get_guest_by_email(conn, auth.user_id()).await?
        .ok_or_else(|| AppError::Forbidden("Guest not found".into()))?;

    let perms = &guest.permissions;

    match request {
        RpcRequest::GetAllDeviceStates if perms.monitor => Ok(()),
        RpcRequest::ExecuteSmartCapability { capability, .. } => {
            match capability {
                DeviceCapability::Lock | DeviceCapability::Unlock 
                    if perms.front_door_and_camera => Ok(()),
                DeviceCapability::SetHeatingSetpoint { .. } 
                    if perms.thermostat_control => Ok(()),
                _ => Err(AppError::Forbidden("Permission denied".into())),
            }
        }
        _ => Err(AppError::Forbidden("Permission denied".into())),
    }
}

Side Effects Processing

When device events occur, the DO processes side effects:

// crates/bf_user/src/notifications.rs

pub struct SideEffectContext<'a> {
    pub env: &'a Env,
    pub conn: &'a mut DOConnection,
    pub property_id: &'a str,
    pub property_address: &'a str,
    pub broadcast_notification: Box<dyn Fn(&Notification)>,
    pub broadcast_pin_code_status: Box<dyn Fn(&str, &PinCodeStatus)>,
}

pub async fn process_side_effects(
    ctx: &mut SideEffectContext<'_>,
    event_info: &SmartEventInfo,
) -> UserResult<()> {
    match &event_info.event {
        SmartEvent::LockStatusChanged { locked, method, user_id } => {
            // Create access log
            let access_log = create_access_log(ctx.conn, &AccessLogRequest {
                device_id: event_info.device_id.clone(),
                device_name: event_info.device_name.clone(),
                event_type: if *locked { "lock" } else { "unlock" },
                method: method.clone(),
                user_identifier: user_id.clone(),
            }).await?;

            // Create notification (uses device name for multi-lock context)
            let notification = create_notification(ctx.conn, &CreateNotificationRequest {
                notification_type: NotificationType::DoorAccess,
                title: format!("{} {}", event_info.device_name,
                    if *locked { "Locked" } else { "Unlocked" }),
                body: Some(format!("Via {}", method.as_deref().unwrap_or("unknown"))),
                device_id: Some(event_info.device_id.clone()),
                device_name: Some(event_info.device_name.clone()),
                action_url: Some(format!("/smart-lock?tab=history&lockId={}",
                    event_info.device_id)),
            }).await?;

            // Broadcast to WebSocket clients
            (ctx.broadcast_notification)(&notification);

            // Send FCM push notifications
            send_push_to_property_users(ctx, &notification).await?;
        }

        SmartEvent::PinCodeManaged { state, partner_user_id } => {
            // Update PIN code status in DB
            if let Some(pin_id) = partner_user_id {
                update_pin_code_yale_state(ctx.conn, pin_id, state).await?;
                (ctx.broadcast_pin_code_status)(pin_id, state);
            }
        }

        // ... other event handlers
    }
    Ok(())
}

DO Alarm API

Used for scheduled tasks like token refresh:

impl UserDurableObject {
    async fn ensure_alarm_scheduled(&self) {
        // Check if credentials exist that need refresh
        if let Ok(Some(creds)) = get_credentials(&self.conn).await {
            if creds.needs_refresh() {
                let delay_ms = creds.time_until_refresh() * 1000;
                self.state.storage().set_alarm(delay_ms).await.ok();
            }
        }
    }

    async fn handle_alarm(&self) -> worker::Result<Response> {
        // Refresh OAuth tokens
        match self.refresh_tokens().await {
            Ok(expires_in) => {
                // Schedule next refresh
                let next_refresh = (expires_in - 300) * 1000; // 5 min before expiry
                self.state.storage().set_alarm(next_refresh).await?;
            }
            Err(e) => {
                // Exponential backoff: 1h, 2h, 4h, 8h, 16h
                let attempts = self.get_refresh_attempts().await;
                let backoff = 3600 * 1000 * (1 << attempts.min(4));
                self.state.storage().set_alarm(backoff).await?;
            }
        }
        Response::ok("Alarm processed")
    }
}

Data Isolation Guarantees

1. Namespace Isolation: Each property_id maps to a unique DO instance
2. Storage Isolation: Each DO has its own SQLite database
3. No Cross-Property Access: RPC requests are scoped to single property
4. Permission Enforcement: Auth checked against property-specific tenant/guest lists
5. Automatic Cleanup: Move-out clears all tenant data

See Also

• sqlx-d1-patterns.md - Database patterns
• workers-architecture.md - Worker overview
• permissions.md - Permission details
• move-in-move-out.md - Data lifecycle