rovenue/

identity.rs

use std::sync::{Arc, Mutex};

use crate::cache::identity::{IdentityRepo, IdentityRow};
use crate::cache::CacheStore;
use crate::error::{RovenueError, RovenueResult};
use crate::observer::{ChangeEvent, ObserverBus};
use crate::time::Clock;

/// A pending identify bind read as ONE pair from the persisted row: the
/// `rovenue_id` the row linked to `app_user_id`. A bind must post this pair,
/// never the row's `app_user_id` with a separately read `rovenue_id`, or a
/// `log_out` in between labels the new anonymous device as the old user.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingBind {
    pub rovenue_id: String,
    pub app_user_id: String,
}

#[derive(Debug, Clone)]
pub struct User {
    pub rovenue_id: String,
    pub app_user_id: Option<String>,
}

pub struct IdentityManager {
    store: Arc<CacheStore>,
    bus: Arc<ObserverBus>,
    clock: Arc<dyn Clock>,
    cached: Mutex<User>,
}

impl IdentityManager {
    pub fn new(store: Arc<CacheStore>, bus: Arc<ObserverBus>, clock: Arc<dyn Clock>) -> Self {
        let repo = IdentityRepo::new(&store);
        let row = match repo.load().ok().flatten() {
            Some(r) => r,
            None => {
                let new_row = IdentityRow {
                    rovenue_id: format!("rov_{}", cuid2::create_id()),
                    app_user_id: None,
                    synced: true,
                    created_at_ms: clock.now_unix_ms(),
                };
                repo.save(&new_row).expect("persist initial identity");
                new_row
            }
        };
        let user = User {
            rovenue_id: row.rovenue_id,
            app_user_id: row.app_user_id,
        };
        Self {
            store,
            bus,
            clock,
            cached: Mutex::new(user),
        }
    }

    pub fn current_user(&self) -> User {
        self.cached.lock().expect("identity mutex poisoned").clone()
    }

    /// The user scope used by the **local cache layer ONLY** — `app_user_id` if
    /// identified, else `rovenue_id`. So a logged-in user's cached entitlements
    /// live in their own namespace and flip on identify/log_out.
    ///
    /// NEVER send this on the wire. Every server endpoint resolves the inbound
    /// identity header/field as a `rovenueId` (the `app_user_id` is a
    /// server-side label attached via `POST /v1/identify`). Use
    /// [`rovenue_id`](Self::rovenue_id) as the network identity instead;
    /// sending the `app_user_id` routes requests to an orphan subscriber row.
    pub fn current_user_scope(&self) -> String {
        let u = self.cached.lock().expect("identity mutex poisoned");
        u.app_user_id
            .clone()
            .unwrap_or_else(|| u.rovenue_id.clone())
    }

    /// Optimistic-local identity set. Persists the `app_user_id` with
    /// `synced: false` (a follow-up `POST /v1/identify` flips it to synced via
    /// [`mark_synced`]). Emits `IdentityChanged` only when the value actually
    /// changes. Returns `true` when the identity changed, `false` when it was
    /// already this `app_user_id` (no write, no emit).
    pub fn set_app_user_id(&self, app_user_id: String) -> RovenueResult<bool> {
        if app_user_id.trim().is_empty() {
            return Err(RovenueError::InvalidArgument()); // dedicated invalid-argument error (distinct from InvalidApiKey)
        }
        // The persist happens inside the identity lock, as in `log_out`, so
        // the persisted row and the in-memory user never disagree: a
        // concurrent `log_out()` cannot land between the cache write and the
        // row write and leave a row that pairs this id with a stale device.
        let changed = {
            let mut u = self.cached.lock().expect("identity mutex poisoned");
            if u.app_user_id.as_deref() == Some(app_user_id.as_str()) {
                false
            } else {
                let row = IdentityRow {
                    rovenue_id: u.rovenue_id.clone(),
                    app_user_id: Some(app_user_id.clone()),
                    synced: false,
                    created_at_ms: self.clock.now_unix_ms(),
                };
                IdentityRepo::new(&self.store).save(&row)?;
                u.app_user_id = Some(app_user_id);
                true
            }
        };
        if changed {
            self.bus.emit(ChangeEvent::IdentityChanged);
        }
        Ok(changed)
    }

    /// Marks the persisted identity row as synced (server `POST /v1/identify`
    /// succeeded) — but ONLY when the row still carries the pair that was
    /// actually sent. This guards two races: `identify(A)`'s slow POST
    /// completing after `identify(B)` overwrote the row (stamping `B` synced
    /// though only `A` was sent), and an old device's POST completing after
    /// `log_out` + `identify(A)` paired `A` with a new device (stamping the new
    /// pair synced though only the old one was sent). Runs under the identity
    /// lock so it cannot write back a row `log_out` just replaced. Does not
    /// touch the in-memory `User` or emit.
    pub fn mark_synced(&self, sent: &PendingBind) -> RovenueResult<()> {
        let _u = self.cached.lock().expect("identity mutex poisoned");
        let repo = IdentityRepo::new(&self.store);
        if let Some(mut row) = repo.load()? {
            if !row.synced
                && row.rovenue_id == sent.rovenue_id
                && row.app_user_id.as_deref() == Some(sent.app_user_id.as_str())
            {
                row.synced = true;
                repo.save(&row)?;
            }
        }
        Ok(())
    }

    /// The pending identify bind, read as one pair from the persisted row
    /// under the identity lock — `Some` only while the row has
    /// `synced == false`, carries an `app_user_id`, and still matches the
    /// in-memory identity. A row the current identity no longer matches (the
    /// identity changed since it was written) yields `None`: the bind is
    /// skipped rather than sent for a device that has moved on.
    pub fn pending_bind(&self) -> Option<PendingBind> {
        let u = self.cached.lock().expect("identity mutex poisoned");
        let row = IdentityRepo::new(&self.store).load().ok().flatten()?;
        if row.synced || row.rovenue_id != u.rovenue_id || row.app_user_id != u.app_user_id {
            return None;
        }
        Some(PendingBind {
            rovenue_id: row.rovenue_id,
            app_user_id: row.app_user_id?,
        })
    }

    /// Returns the `app_user_id` awaiting server sync — the `app_user_id` of
    /// [`pending_bind`](Self::pending_bind).
    pub fn pending_app_user_id(&self) -> Option<String> {
        self.pending_bind().map(|b| b.app_user_id)
    }

    /// The current anonymous device id.
    pub fn rovenue_id(&self) -> String {
        self.cached
            .lock()
            .expect("identity mutex poisoned")
            .rovenue_id
            .clone()
    }

    /// Resets identity to a fresh anonymous user: mints a new `rovenue_id`,
    /// drops any `app_user_id`, persists, and notifies observers. Callers that
    /// hold scope-bound caches must clear them after this returns.
    pub fn log_out(&self) -> RovenueResult<()> {
        let new_row = IdentityRow {
            rovenue_id: format!("rov_{}", cuid2::create_id()),
            app_user_id: None,
            synced: true,
            created_at_ms: self.clock.now_unix_ms(),
        };
        {
            // Row and cache swap inside one critical section, so a pending
            // bind read under the same lock sees either the old pair or no
            // pending bind — never the old app_user_id with the new device.
            let mut u = self.cached.lock().expect("identity mutex poisoned");
            IdentityRepo::new(&self.store).save(&new_row)?;
            u.rovenue_id = new_row.rovenue_id.clone();
            u.app_user_id = None;
        }
        self.bus.emit(ChangeEvent::IdentityChanged);
        Ok(())
    }
}