diff --git a/.coderabbit.yaml b/.coderabbit.yaml new file mode 100644 index 0000000..84b066f --- /dev/null +++ b/.coderabbit.yaml @@ -0,0 +1,253 @@ +# yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json +# +# qMonitor review policy. Matches GitHub Actions: +# main → canary prerelease v{package.json}-canary.{run_number} +# release → stable GitHub Release v{package.json} (immutable) +# PRs → CI only (no artifacts); targets main | master | release +# +language: en-US +tone_instructions: >- + Concise reviews for this Tauri 2 + React desktop app. Prefer merge-blocking + bugs, security, data loss, and release-channel mistakes over style nits. Do + not invent a second version source besides package.json. + +reviews: + profile: chill + request_changes_workflow: false + high_level_summary: true + high_level_summary_in_walkthrough: true + poem: false + collapse_walkthrough: true + sequence_diagrams: true + review_status: true + review_details: true + review_progress: true + abort_on_close: true + auto_apply_labels: false + auto_assign_reviewers: false + + labeling_instructions: + - label: rust + instructions: Changes under src-tauri/ (Rust core, Tauri commands, Cargo). + - label: ui + instructions: Changes under src/ (React / Vite frontend). + - label: ci + instructions: Changes under .github/ or scripts/ that affect CI, canary, or release. + - label: packaging + instructions: Installer, bundle, or distro packaging (packaging/, tauri.conf.json bundle targets). + - label: release + instructions: Version bumps, GitHub Release/canary workflow, or update-channel logic. + - label: security + instructions: Auth, OAuth/PKCE, keyring, tokens, webhooks, or process-identity trust. + + path_filters: + - "!pnpm-lock.yaml" + - "!**/Cargo.lock" + - "!src-tauri/target/**" + - "!src-tauri/gen/**" + - "!dist/**" + - "!icons/**" + - "!public/favicon.svg" + - "!**/*.png" + - "!**/*.ico" + - "!**/*.icns" + + path_instructions: + - path: ".github/workflows/**" + instructions: | + Branch and publish model (do not invent new channels): + - Push to `main` runs canary.yml: immutable prerelease tag + `v{package.json.version}-canary.{GITHUB_RUN_NUMBER}`, prerelease=true, + make_latest=false. Never reuse or delete unique v*-canary.* tags. + The old rolling `canary` tag cleanup is one-shot leftover removal only. + - Push to `release` runs release.yml: immutable stable tag `v{package.json.version}`. + package.json is the sole version source. If that tag already exists, fail — + never mutate a shipped release. Cargo.toml / tauri.conf.json are synced at + build via scripts/sync-version.mjs, not edited by hand as source of truth. + - PRs targeting main, master, or release run ci.yml (frontend build + cargo test) + only. They must not publish GitHub Releases or upload installer artifacts. + - build-artifacts.yml is reusable: optional version override for canary; + Windows NSIS+MSI, Linux AppImage+.deb, then Arch .pkg.tar.zst from the .deb. + Flag secrets in logs, unpinned privileged actions, and concurrency that could + cancel an in-flight stable release (release concurrency cancel-in-progress is false). + + - path: "src-tauri/src/{auth,pkce,oauth_loopback,device}.rs" + instructions: | + Security-sensitive: device login (auth code + PKCE), loopback redirect, OS keyring. + Tokens (access_token, session_token/refresh) must never be logged, written to + config JSON, or returned to the UI except as booleans (has_access_token). + Preserve PKCE verifier/state binding, device_id from hashed install salt, and + 401 → refresh with device_id + refresh token then retry. Flag SSRF, open + redirects, and storing secrets outside the keyring. + + - path: "src-tauri/src/{push,persist,db,session,live_session}.rs" + instructions: | + Outbox path: completed sessions → local Turso → POST {apiRoot}/webhooks/qmonitor + with Bearer access token. HTTP 2xx acks the row; 401 refreshes; other failures + retry. Do not drop unacked rows. Retention purge is 7 or 30 days for acked rows + only. Payload schema_version, session_id, steam_app_id, timestamps, duration + must stay compatible with Questory. Flag double-push, lost ack, and clock/duration + inconsistencies. + + - path: "src-tauri/src/{identity,detect}/**/*.rs" + instructions: | + Steam-first identity: AppID via launch reaper + local library, then Discord + detectable catalog, then local catalog + user confirm. Deny-list (identity/deny.rs) + must never treat launchers, overlays, browsers, or crash handlers as games. + Flag false-positive game detection, ignoring is_denied, and path/exe fingerprint + collisions that could mix two titles. + + - path: "src-tauri/src/{update_check,config}.rs" + instructions: | + Updates are notify-only (link to GitHub Releases), never auto-install. + Stable channel uses GitHub "latest" and must ignore prerelease / *-canary.* tags. + Canary channel picks the newest prerelease whose tag matches vX.Y.Z-canary.N + (not a rolling `canary` tag). open_release_url must stay allowlisted to + github.com/Questory-Labs/qMonitor/releases. Default update_channel is stable. + Config lives under the OS config dir (qMonitor); do not write tokens there. + + - path: "src-tauri/**/*.rs" + instructions: | + Tauri 2 + Tokio. Prefer existing modules over new crates. Tauri commands stay + thin; keep process polling, identity, and DB off the UI thread. Tests are + #[cfg(test)] in the same module (cargo test in src-tauri) — new logic needs + coverage for success, error, and platform #cfg paths. Do not suggest unwrap + on fallible I/O in production paths. + + - path: "src/**/*.{ts,tsx,css}" + instructions: | + Compact tray-friendly UI (Home / Games / Settings). Invoke Tauri commands + through the existing helpers (including invokeTimeout). Do not call Questory + HTTP from the renderer — auth and webhook live in Rust. UpdateSettings channel + is "stable" | "canary" only. Avoid leaking tokens into React state beyond + AuthState booleans. + + - path: "{package.json,src-tauri/tauri.conf.json,src-tauri/Cargo.toml,scripts/sync-version.mjs}" + instructions: | + package.json.version is the only version source. sync-version.mjs copies it to + Cargo.toml and tauri.conf.json at build; canary semver needs wix.version set to + the core X.Y.Z because MSI ProductVersion cannot encode -canary.N. Do not + introduce a second version field or bump Cargo/tauri independently. + + - path: "packaging/**" + instructions: | + Arch package is a CI repack of the official .deb (PKGBUILD source rewritten + in build-artifacts.yml). Keep runtime depends (GTK/WebKit) accurate. Do not + assume an AUR source URL unless the PR is explicitly adding one. + + auto_review: + enabled: true + drafts: false + auto_incremental_review: true + auto_pause_after_reviewed_commits: 5 + # Default branch (main) is always reviewed. Also cover stable + legacy CI targets. + base_branches: + - "^release$" + - "^master$" + ignore_title_keywords: + - WIP + - "[skip review]" + - "[skip cr]" + labels: + - "!do-not-review" + ignore_usernames: + - dependabot[bot] + - renovate[bot] + - github-actions[bot] + + pre_merge_checks: + docstrings: + mode: off + title: + mode: warning + requirements: >- + Concise imperative summary of the change (e.g. "Fix Steam reaper AppID + resolution"). Do not put version tags or "canary"/"release" in the title + unless the PR actually changes publish/update-channel behavior. + description: + mode: warning + issue_assessment: + mode: off + custom_checks: + - name: Release vs canary versioning + mode: warning + instructions: >- + Fail if a PR targeting `release` sets package.json version to a + prerelease (contains -canary or other -prerelease) or reuses an + existing v* GitHub Release tag. Fail if workflows publish artifacts + from pull_request events, retag a shipped vX.Y.Z, or treat a rolling + `canary` tag as the canary channel (channel is unique + v{version}-canary.{run_number} prereleases). Pass when versioning is + unchanged or correctly follows package.json as the sole source. + + tools: + clippy: + enabled: true + actionlint: + enabled: true + zizmor: + enabled: true + gitleaks: + enabled: true + trufflehog: + enabled: true + github-checks: + enabled: true + shellcheck: + enabled: true + yamllint: + enabled: true + markdownlint: + enabled: true + oxc: + enabled: true + osv-scanner: + enabled: true + +chat: + auto_reply: true + art: false + +knowledge_base: + opt_out: false + web_search: + enabled: true + code_guidelines: + enabled: true + filePatterns: + - files: README.md + applyTo: "**/*" + learnings: + scope: local + issues: + scope: local + pull_requests: + scope: local + linked_repositories: + - repository: Questory-Labs/Questory + instructions: >- + Questory API/web that qMonitor authenticates against and pushes to. + Device OAuth lives at /oauth/qmonitor/{authorize,token,revoke}. + Session ingest is POST /webhooks/qmonitor (proxied as /api/webhooks/qmonitor + from the web origin). Flag payload or auth contract drift against that repo. + +code_generation: + docstrings: + language: en-US + path_instructions: + - path: "src-tauri/**/*.rs" + instructions: >- + Use rustdoc on exported types and Tauri commands. Document invariants, + error cases, and why — not what the code already says. Skip trivial + private helpers. + unit_tests: + path_instructions: + - path: "src-tauri/**/*.rs" + instructions: >- + Add #[cfg(test)] tests in the same module. Cover identity resolution, + deny-list, outbox ack/retry, update-channel tag parsing, and auth token + handling without hitting the network (mock HTTP / temp dirs). + - path: "src/**/*.{ts,tsx}" + instructions: >- + Only add frontend tests if the repo already has a runner for them. + Prefer covering logic in Rust when the behavior is enforced there. diff --git a/.github/actions/setup-rust/action.yml b/.github/actions/setup-rust/action.yml new file mode 100644 index 0000000..7159a03 --- /dev/null +++ b/.github/actions/setup-rust/action.yml @@ -0,0 +1,17 @@ +name: Setup Rust +description: Install stable Rust and share the src-tauri cargo cache across CI, canary, and release. + +inputs: + platform: + description: rust-cache shared-key (linux or windows) + required: true + +runs: + using: composite + steps: + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + with: + workspaces: src-tauri + shared-key: ${{ inputs.platform }} + save-if: ${{ github.ref == 'refs/heads/main' || github.ref == 'refs/heads/release' }} diff --git a/.github/workflows/build-artifacts.yml b/.github/workflows/build-artifacts.yml index 811e34f..ace7d0a 100644 --- a/.github/workflows/build-artifacts.yml +++ b/.github/workflows/build-artifacts.yml @@ -83,12 +83,9 @@ jobs: node-version: 22 cache: pnpm - - uses: dtolnay/rust-toolchain@stable - - - uses: Swatinem/rust-cache@v2 + - uses: ./.github/actions/setup-rust with: - workspaces: src-tauri - shared-key: ${{ matrix.platform }}-${{ needs.prepare.outputs.version }} + platform: ${{ matrix.platform }} - run: pnpm install --frozen-lockfile diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 159ee56..db01870 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,7 +12,11 @@ jobs: test: strategy: matrix: - os: [windows-latest, ubuntu-22.04] + include: + - os: windows-latest + platform: windows + - os: ubuntu-22.04 + platform: linux runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v7 @@ -31,10 +35,9 @@ jobs: with: node-version: 22 cache: pnpm - - uses: dtolnay/rust-toolchain@stable - - uses: Swatinem/rust-cache@v2 + - uses: ./.github/actions/setup-rust with: - workspaces: src-tauri + platform: ${{ matrix.platform }} - run: pnpm install --frozen-lockfile - run: pnpm build - name: Cargo test diff --git a/package.json b/package.json index df965f0..759d3ba 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "qmonitor", "private": true, - "version": "0.0.1", + "version": "0.0.2", "type": "module", "scripts": { "dev": "vite", @@ -15,6 +15,7 @@ "dependencies": { "@tauri-apps/api": "^2", "@tauri-apps/plugin-autostart": "^2.5.1", + "@tauri-apps/plugin-dialog": "^2.7.2", "@tauri-apps/plugin-opener": "^2", "@tauri-apps/plugin-shell": "^2.3.5", "react": "^19.1.0", @@ -29,6 +30,8 @@ "vite": "^7.0.4" }, "pnpm": { - "onlyBuiltDependencies": ["esbuild"] + "onlyBuiltDependencies": [ + "esbuild" + ] } } \ No newline at end of file diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 22d63c1..319b927 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -14,6 +14,9 @@ importers: '@tauri-apps/plugin-autostart': specifier: ^2.5.1 version: 2.5.1 + '@tauri-apps/plugin-dialog': + specifier: ^2.7.2 + version: 2.7.2 '@tauri-apps/plugin-opener': specifier: ^2 version: 2.5.4 @@ -533,6 +536,9 @@ packages: '@tauri-apps/plugin-autostart@2.5.1': resolution: {integrity: sha512-zS/xx7yzveCcotkA+8TqkI2lysmG2wvQXv2HGAVExITmnFfHAdj1arGsbbfs3o6EktRHf6l34pJxc3YGG2mg7w==} + '@tauri-apps/plugin-dialog@2.7.2': + resolution: {integrity: sha512-pX0IGm1I3I6wc+zeKYcq1GSqogK6okCNX5fOdaNU5ab1AjGS6l1E5wFNjEb7meg7ZFSp0JUs+0jQGQNyOvLrsg==} + '@tauri-apps/plugin-opener@2.5.4': resolution: {integrity: sha512-1HnPkb+AmgO29HBazm4uPLKB+r7zzcTBW1d0fyYp1uP+jwtpoiNDGKMMzz58SFp49nOIrxdE3aUJtT57lfO9CQ==} @@ -1096,6 +1102,10 @@ snapshots: dependencies: '@tauri-apps/api': 2.11.1 + '@tauri-apps/plugin-dialog@2.7.2': + dependencies: + '@tauri-apps/api': 2.11.1 + '@tauri-apps/plugin-opener@2.5.4': dependencies: '@tauri-apps/api': 2.11.1 diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 5b7a8fb..0896ef4 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -3169,6 +3169,7 @@ checksum = "e3e0adef53c21f888deb4fa59fc59f7eb17404926ee8a6f59f5df0fd7f9f3272" dependencies = [ "bitflags 2.13.1", "block2", + "libc", "objc2", "objc2-core-foundation", ] @@ -3672,7 +3673,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.12.1", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.119", @@ -3698,12 +3699,14 @@ dependencies = [ "tauri", "tauri-build", "tauri-plugin-autostart", + "tauri-plugin-dialog", "tauri-plugin-opener", "tauri-plugin-shell", "tempfile", "thiserror 2.0.20", "tokio", "tracing", + "tracing-appender", "tracing-subscriber", "turso", "url", @@ -4083,6 +4086,30 @@ dependencies = [ "web-sys", ] +[[package]] +name = "rfd" +version = "0.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a15ad77d9e70a92437d8f74c35d99b4e4691128df018833e99f90bcd36152672" +dependencies = [ + "block2", + "dispatch2", + "glib-sys", + "gobject-sys", + "gtk-sys", + "js-sys", + "log", + "objc2", + "objc2-app-kit", + "objc2-core-foundation", + "objc2-foundation", + "raw-window-handle", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "windows-sys 0.60.2", +] + [[package]] name = "ring" version = "0.17.14" @@ -5249,6 +5276,48 @@ dependencies = [ "thiserror 2.0.20", ] +[[package]] +name = "tauri-plugin-dialog" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2d3c1dbe38037e7f590cdf2492594d5ceebe031e7bc7e827509b22a999d2940" +dependencies = [ + "log", + "raw-window-handle", + "rfd", + "serde", + "serde_json", + "tauri", + "tauri-plugin", + "tauri-plugin-fs", + "thiserror 2.0.20", + "url", +] + +[[package]] +name = "tauri-plugin-fs" +version = "2.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7ecc274121aca0c036a2b42d1cbe83d368d348f54e0bb8a735c2b1548e8f371" +dependencies = [ + "anyhow", + "dunce", + "glob", + "log", + "objc2-foundation", + "percent-encoding", + "schemars 0.8.22", + "serde", + "serde_json", + "serde_repr", + "tauri", + "tauri-plugin", + "tauri-utils", + "thiserror 2.0.20", + "toml 1.1.4+spec-1.1.0", + "url", +] + [[package]] name = "tauri-plugin-opener" version = "2.5.4" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 3dc3523..c8d2daf 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -18,6 +18,7 @@ tauri = { version = "2", features = ["tray-icon"] } tauri-plugin-opener = "2" tauri-plugin-shell = "2" tauri-plugin-autostart = "2" +tauri-plugin-dialog = "2" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time"] } @@ -28,6 +29,7 @@ chrono = { version = "0.4", features = ["serde"] } base64 = "0.22" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } +tracing-appender = "0.2" sysinfo = "0.33" keyring = { version = "3", features = ["apple-native", "windows-native", "linux-native"] } sha2 = "0.10" diff --git a/src-tauri/capabilities/default.json b/src-tauri/capabilities/default.json index 5c88494..625979d 100644 --- a/src-tauri/capabilities/default.json +++ b/src-tauri/capabilities/default.json @@ -6,6 +6,7 @@ "permissions": [ "core:default", "opener:default", + "dialog:allow-open", "shell:allow-open", "autostart:allow-enable", "autostart:allow-disable", diff --git a/src-tauri/src/auth.rs b/src-tauri/src/auth.rs index dbe8793..2502ccc 100644 --- a/src-tauri/src/auth.rs +++ b/src-tauri/src/auth.rs @@ -1,7 +1,10 @@ //! Device login token storage, health detect, and OAuth token calls. +use std::sync::{Mutex, OnceLock}; + use keyring::Entry; use serde::Deserialize; +use tokio::sync::Mutex as AsyncMutex; use crate::config::{ AppConfig, DetectedService, KEYRING_ACCESS, KEYRING_SERVICE, KEYRING_SESSION, @@ -9,6 +12,7 @@ use crate::config::{ use crate::device; use crate::oauth_loopback::REDIRECT_URI; use crate::pkce; +use crate::push::http_client; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] @@ -41,21 +45,97 @@ struct TokenResponse { refresh_token: Option, } +#[derive(Default)] +struct TokenCache { + access: Option, + refresh: Option, + keyring_hydrated: bool, +} + +fn token_cache() -> &'static Mutex { + static CACHE: OnceLock> = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(TokenCache::default())) +} + +fn refresh_lock() -> &'static AsyncMutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| AsyncMutex::new(())) +} + +fn cache_put(access: Option<&str>, refresh: Option<&str>) { + if let Ok(mut c) = token_cache().lock() { + if let Some(a) = access { + c.access = Some(a.to_string()); + } + if let Some(r) = refresh { + c.refresh = Some(r.to_string()); + } + } +} + +fn cache_clear() { + if let Ok(mut c) = token_cache().lock() { + c.access = None; + c.refresh = None; + c.keyring_hydrated = true; + } +} + +fn keyring_get(key: &str) -> Result, String> { + let entry = Entry::new(KEYRING_SERVICE, key).map_err(|e| e.to_string())?; + match entry.get_password() { + Ok(p) if !p.is_empty() => Ok(Some(p)), + Ok(_) => Ok(None), + Err(keyring::Error::NoEntry) => Ok(None), + Err(e) => Err(e.to_string()), + } +} + +fn hydrate_keyring_once(cache: &mut TokenCache) { + if cache.keyring_hydrated { + return; + } + cache.keyring_hydrated = true; + match keyring_get(KEYRING_ACCESS) { + Ok(v) => { + if cache.access.is_none() { + cache.access = v; + } + } + Err(e) => tracing::warn!(%e, "keyring access read failed; keeping cache"), + } + match keyring_get(KEYRING_SESSION) { + Ok(v) => { + if cache.refresh.is_none() { + cache.refresh = v; + } + } + Err(e) => tracing::warn!(%e, "keyring refresh read failed; keeping cache"), + } +} + +fn keyring_set(key: &str, value: &str, what: &str) { + match Entry::new(KEYRING_SERVICE, key) { + Ok(entry) => { + if let Err(e) = entry.set_password(value) { + tracing::warn!(%e, "{what} write failed; token cached in memory"); + } + } + Err(e) => tracing::warn!(%e, "{what} write failed; token cached in memory"), + } +} + pub fn store_tokens(access: &str, refresh: Option<&str>) -> Result<(), String> { - Entry::new(KEYRING_SERVICE, KEYRING_ACCESS) - .map_err(|e| e.to_string())? - .set_password(access) - .map_err(|e| e.to_string())?; + cache_put(Some(access), refresh); + keyring_set(KEYRING_ACCESS, access, "keyring access"); if let Some(s) = refresh { - Entry::new(KEYRING_SERVICE, KEYRING_SESSION) - .map_err(|e| e.to_string())? - .set_password(s) - .map_err(|e| e.to_string())?; + keyring_set(KEYRING_SESSION, s, "keyring refresh"); } Ok(()) } pub fn clear_tokens() -> Result<(), String> { + cache_clear(); let _ = Entry::new(KEYRING_SERVICE, KEYRING_ACCESS) .ok() .and_then(|e| e.delete_credential().ok()); @@ -71,15 +151,15 @@ pub fn get_access_token(cfg: &AppConfig) -> Option { return Some(dev.clone()); } } - Entry::new(KEYRING_SERVICE, KEYRING_ACCESS) - .ok() - .and_then(|e| e.get_password().ok()) + let mut cache = token_cache().lock().ok()?; + hydrate_keyring_once(&mut cache); + cache.access.clone() } pub fn get_refresh_token() -> Option { - Entry::new(KEYRING_SERVICE, KEYRING_SESSION) - .ok() - .and_then(|e| e.get_password().ok()) + let mut cache = token_cache().lock().ok()?; + hydrate_keyring_once(&mut cache); + cache.refresh.clone() } pub fn auth_state(cfg: &AppConfig) -> Option { @@ -158,7 +238,7 @@ pub async fn exchange_authorization_code( return Err("state mismatch".into()); } let token_url = cfg.token_url().ok_or("api root not configured")?; - let client = reqwest::Client::new(); + let client = http_client(); let res = client .post(&token_url) .json(&serde_json::json!({ @@ -183,10 +263,11 @@ pub async fn exchange_authorization_code( } pub async fn refresh_access_token(cfg: &AppConfig) -> Result { + let _guard = refresh_lock().lock().await; let refresh = get_refresh_token().ok_or("no refresh token")?; let device_id = device::device_id()?; let token_url = cfg.token_url().ok_or("api root not configured")?; - let client = reqwest::Client::new(); + let client = http_client(); let res = client .post(&token_url) .json(&serde_json::json!({ @@ -215,7 +296,7 @@ pub async fn revoke_remote(cfg: &AppConfig) -> Result<(), String> { let Some(token) = token else { return Ok(()); }; - let client = reqwest::Client::new(); + let client = http_client(); let mut body = serde_json::json!({ "token": token, "client_id": "qmonitor", @@ -275,4 +356,20 @@ mod tests { assert_eq!(c, "abc"); assert_eq!(s, "xyz"); } + + #[test] + fn cache_survives_without_keyring_and_clears() { + cache_put(Some("access-1"), Some("refresh-1")); + let cfg = AppConfig::default(); + assert_eq!(get_access_token(&cfg).as_deref(), Some("access-1")); + assert_eq!(get_refresh_token().as_deref(), Some("refresh-1")); + cache_clear(); + // Hydrated empty cache: no keyring token in CI. + let mut c = token_cache().lock().unwrap(); + c.keyring_hydrated = true; + c.access = None; + c.refresh = None; + drop(c); + assert!(get_access_token(&cfg).is_none()); + } } diff --git a/src-tauri/src/config.rs b/src-tauri/src/config.rs index ea4cc9a..f3e5ac5 100644 --- a/src-tauri/src/config.rs +++ b/src-tauri/src/config.rs @@ -21,6 +21,33 @@ pub enum DetectedService { Be, } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)] +#[serde(rename_all = "lowercase")] +pub enum LogLevel { + #[default] + Off, + Error, + Warn, + Info, + Debug, +} + +impl LogLevel { + pub fn env_filter(self) -> &'static str { + match self { + Self::Off => "off", + Self::Error => "qmonitor_lib=error,qmonitor=error", + Self::Warn => "qmonitor_lib=warn,qmonitor=warn", + Self::Info => "qmonitor_lib=info,qmonitor=info", + Self::Debug => "qmonitor_lib=debug,qmonitor=debug", + } + } + + pub fn file_enabled(self) -> bool { + self != Self::Off + } +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)] #[serde(rename_all = "lowercase")] pub enum UpdateChannel { @@ -56,6 +83,9 @@ pub struct AppConfig { /// GitHub release channel to poll for updates (stable = Latest, canary = newest prerelease). #[serde(default)] pub update_channel: UpdateChannel, + /// File log verbosity. Default off — no log files. + #[serde(default)] + pub log_level: LogLevel, /// Dev fallback when device login is unavailable. pub dev_access_token: Option, } @@ -77,6 +107,7 @@ impl Default for AppConfig { minimize_to_tray: false, close_to_tray: false, update_channel: UpdateChannel::Stable, + log_level: LogLevel::Off, dev_access_token: None, } } @@ -184,6 +215,17 @@ mod tests { assert_eq!(cfg.update_channel, UpdateChannel::Stable); assert!(!cfg.minimize_to_tray); assert!(!cfg.close_to_tray); + assert_eq!(cfg.log_level, LogLevel::Off); + } + + #[test] + fn log_level_roundtrip() { + let mut cfg = AppConfig::default(); + cfg.log_level = LogLevel::Debug; + let raw = serde_json::to_string(&cfg).expect("ser"); + let back: AppConfig = serde_json::from_str(&raw).expect("de"); + assert_eq!(back.log_level, LogLevel::Debug); + assert!(raw.contains("\"logLevel\":\"debug\"")); } #[test] diff --git a/src-tauri/src/db.rs b/src-tauri/src/db.rs index 2905221..89bf993 100644 --- a/src-tauri/src/db.rs +++ b/src-tauri/src/db.rs @@ -57,8 +57,8 @@ pub struct SessionRow { } pub struct TursoDb { - #[allow(dead_code)] - db: Database, + /// Held so the connection is not dropped with the builder handle. + _db: Database, conn: Connection, } @@ -74,7 +74,7 @@ impl TursoDb { .await .map_err(|e| format!("turso open: {e}"))?; let conn = db.connect().map_err(|e| format!("turso conn: {e}"))?; - let this = Self { db, conn }; + let this = Self { _db: db, conn }; this.migrate().await?; Ok(this) } @@ -126,8 +126,12 @@ impl TursoDb { Ok(()) } - pub async fn open_session(&self, identity: &GameIdentity) -> Result { - // Never create a second active for the same identity (restart / race safety). + /// Open (or reuse) an active row. Existing DB `started_at` wins over `started_at`. + pub async fn open_session_at( + &self, + identity: &GameIdentity, + started_at: DateTime, + ) -> Result { if let Some(existing) = self .list_active() .await? @@ -145,7 +149,7 @@ impl TursoDb { steam_app_id: identity.steam_app_id, exe: identity.exe.clone(), source: identity.source.clone(), - started_at: Utc::now(), + started_at, ended_at: None, duration_secs: None, push_status: PushStatus::Active, @@ -253,17 +257,29 @@ impl TursoDb { Ok(()) } - pub async fn end_session(&self, id: &str) -> Result { + pub async fn end_session_at( + &self, + id: &str, + ended_at: DateTime, + ) -> Result { let mut row = self .get_session(id) .await? .ok_or_else(|| "missing".to_string())?; - let ended = Utc::now(); + if row.push_status != PushStatus::Active { + return Ok(row); + } + let ended = if ended_at < row.started_at { + row.started_at + } else { + ended_at + }; let duration = (ended - row.started_at).num_seconds().max(0); + let next_retry_at = Utc::now(); row.ended_at = Some(ended); row.duration_secs = Some(duration); row.push_status = PushStatus::Pending; - row.next_retry_at = Some(Utc::now()); + row.next_retry_at = Some(next_retry_at); self.conn .execute( r#"UPDATE sessions SET ended_at=?, duration_secs=?, push_status=?, next_retry_at=? WHERE id=?"#, @@ -271,7 +287,7 @@ impl TursoDb { ended.to_rfc3339(), duration, PushStatus::Pending.as_str(), - Utc::now().to_rfc3339(), + next_retry_at.to_rfc3339(), id, ), ) @@ -637,9 +653,9 @@ mod tests { let path = dir.path().join("test.db"); let db = TursoDb::open(&path).await.expect("open"); db.ping().await.expect("ping"); - let row = db.open_session(&sample_identity()).await.unwrap(); + let row = db.open_session_at(&sample_identity(), Utc::now()).await.unwrap(); assert_eq!(row.push_status, PushStatus::Active); - let ended = db.end_session(&row.id).await.unwrap(); + let ended = db.end_session_at(&row.id, Utc::now()).await.unwrap(); assert_eq!(ended.push_status, PushStatus::Pending); db.mark_synced(&row.id).await.unwrap(); // Force old ack via SQL @@ -662,8 +678,8 @@ mod tests { let dir = tempdir().unwrap(); let path = dir.path().join("dup.db"); let db = TursoDb::open(&path).await.expect("open"); - let a = db.open_session(&sample_identity()).await.unwrap(); - let b = db.open_session(&sample_identity()).await.unwrap(); + let a = db.open_session_at(&sample_identity(), Utc::now()).await.unwrap(); + let b = db.open_session_at(&sample_identity(), Utc::now()).await.unwrap(); assert_eq!(a.id, b.id); assert_eq!(db.list_active().await.unwrap().len(), 1); } @@ -673,7 +689,7 @@ mod tests { let dir = tempdir().unwrap(); let path = dir.path().join("discard.db"); let db = TursoDb::open(&path).await.expect("open"); - let row = db.open_session(&sample_identity()).await.unwrap(); + let row = db.open_session_at(&sample_identity(), Utc::now()).await.unwrap(); db.discard_session(&row.id).await.unwrap(); assert!(db.list_active().await.unwrap().is_empty()); assert!(db.list_due_pushes().await.unwrap().is_empty()); diff --git a/src-tauri/src/health.rs b/src-tauri/src/health.rs new file mode 100644 index 0000000..d8cedd5 --- /dev/null +++ b/src-tauri/src/health.rs @@ -0,0 +1,383 @@ +//! Liveness stamps, tray tooltip, and file logging. + +use std::fs; +use std::io::{self, Write}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Mutex, OnceLock}; +use std::time::{Duration, SystemTime}; + +use chrono::{DateTime, Utc}; +use tracing_appender::non_blocking::{NonBlocking, WorkerGuard}; +use tracing_subscriber::fmt::writer::MakeWriterExt; +use tracing_subscriber::prelude::*; +use tracing_subscriber::{reload, EnvFilter}; + +use crate::config::{AppConfig, LogLevel}; + +/// Delete log files older than this when logging is on. +pub const LOG_KEEP_DAYS: u64 = 3; +/// Cap total size of `qmonitor.log*` files. +pub const LOG_MAX_BYTES: u64 = 5 * 1024 * 1024; +pub const LOG_PRUNE_EVERY: Duration = Duration::from_secs(30 * 60); + +static FILE_SINK: Mutex> = Mutex::new(None); +static FILE_ON: AtomicBool = AtomicBool::new(false); +static FILTER_RELOAD: OnceLock> = + OnceLock::new(); + +#[derive(Debug, Clone, Default)] +pub struct RuntimeHealth { + pub last_detect_at: Option>, + pub last_persist_at: Option>, + pub last_push_at: Option>, + pub db_ok: bool, + pub db_generation: u64, + pub db_reconnects: u64, + pub detect_timeouts: u64, + pub last_error: Option, +} + +impl RuntimeHealth { + pub fn loop_alive(&self, poll_interval_secs: u64) -> bool { + let Some(at) = self.last_detect_at else { + return false; + }; + let max = chrono::Duration::seconds((poll_interval_secs.max(1) * 3) as i64); + Utc::now().signed_duration_since(at) < max + } + + pub fn tray_tooltip(&self, poll_interval_secs: u64) -> String { + let poll = match self.last_detect_at { + Some(at) => { + let secs = Utc::now().signed_duration_since(at).num_seconds().max(0); + format!("poll {secs}s ago") + } + None => "poll —".into(), + }; + let db = if self.db_ok { "DB ok" } else { "DB down" }; + let stuck = if self.loop_alive(poll_interval_secs) { + "" + } else { + " · loop stuck" + }; + format!("qMonitor · {poll} · {db}{stuck}") + } +} + +pub fn log_dir() -> PathBuf { + AppConfig::config_dir().join("logs") +} + +/// Stderr + rolling daily file. File sink is created on first enabled write (default off). +pub fn init_tracing() { + let level = AppConfig::load().log_level; + FILE_ON.store(level.file_enabled(), Ordering::Relaxed); + prune_now(level); + + let (filter, reload_handle) = + reload::Layer::new(EnvFilter::new(level.env_filter())); + let _ = FILTER_RELOAD.set(reload_handle); + + tracing_subscriber::registry() + .with(filter) + .with( + tracing_subscriber::fmt::layer() + .with_writer(std::io::stderr.and(GatedMakeWriter)), + ) + .init(); + + if level.file_enabled() { + tracing::info!(path = %log_dir().display(), level = ?level, "file logging on"); + } +} + +pub fn apply_log_level(level: LogLevel) { + FILE_ON.store(level.file_enabled(), Ordering::Relaxed); + if !level.file_enabled() { + drop_file_sink(); + } + if let Some(handle) = FILTER_RELOAD.get() { + if let Err(e) = handle.reload(EnvFilter::new(level.env_filter())) { + tracing::warn!(%e, "failed to reload log filter"); + } + } + let dir = log_dir(); + if level.file_enabled() { + let _ = fs::create_dir_all(&dir); + } + prune_log_dir(&dir, prune_keep_days(level), prune_max_bytes(level)); +} + +pub fn prune_now(level: LogLevel) { + prune_log_dir(&log_dir(), prune_keep_days(level), prune_max_bytes(level)); +} + +fn prune_keep_days(level: LogLevel) -> u64 { + if level.file_enabled() { + LOG_KEEP_DAYS + } else { + 0 + } +} + +fn prune_max_bytes(level: LogLevel) -> u64 { + if level.file_enabled() { + LOG_MAX_BYTES + } else { + 0 + } +} + +fn is_qmonitor_log(name: &str) -> bool { + name.starts_with("qmonitor.log") +} + +fn active_daily_log_name() -> Option { + let sink_live = FILE_SINK + .lock() + .unwrap_or_else(|e| e.into_inner()) + .is_some(); + if !sink_live { + return None; + } + Some(format!( + "qmonitor.log.{}", + chrono::Local::now().format("%Y-%m-%d") + )) +} + +fn is_protected_log(path: &Path, active_name: Option<&str>) -> bool { + let Some(active) = active_name else { + return false; + }; + path.file_name() + .and_then(|n| n.to_str()) + .is_some_and(|name| name == active) +} + +/// Age + size cap. `keep_days == 0` or `max_bytes == 0` deletes every `qmonitor.log*` file +/// except the active daily file while the sink is using it. +pub fn prune_log_dir(dir: &Path, keep_days: u64, max_bytes: u64) { + let Ok(entries) = fs::read_dir(dir) else { + return; + }; + let active_name = active_daily_log_name(); + let mut files: Vec<(PathBuf, SystemTime, u64)> = Vec::new(); + for entry in entries.flatten() { + let path = entry.path(); + let name = path + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or_default(); + if !is_qmonitor_log(name) { + continue; + } + let Ok(meta) = entry.metadata() else { + continue; + }; + if !meta.is_file() { + continue; + } + let modified = meta.modified().unwrap_or(SystemTime::UNIX_EPOCH); + files.push((path, modified, meta.len())); + } + + if keep_days == 0 || max_bytes == 0 { + for (path, _, _) in files { + if is_protected_log(&path, active_name.as_deref()) { + continue; + } + let _ = fs::remove_file(path); + } + return; + } + + let cutoff = SystemTime::now() + .checked_sub(Duration::from_secs(keep_days.saturating_mul(86_400))) + .unwrap_or(SystemTime::UNIX_EPOCH); + files.retain(|(path, modified, _)| { + if is_protected_log(path, active_name.as_deref()) { + return true; + } + if *modified < cutoff { + let _ = fs::remove_file(path); + false + } else { + true + } + }); + + files.sort_by_key(|(_, modified, _)| *modified); + let mut total: u64 = files.iter().map(|(_, _, len)| *len).sum(); + for (path, _, len) in files { + if total <= max_bytes { + break; + } + if is_protected_log(&path, active_name.as_deref()) { + continue; + } + if fs::remove_file(&path).is_ok() { + total = total.saturating_sub(len); + } + } +} + +fn drop_file_sink() { + if let Ok(mut g) = FILE_SINK.lock() { + *g = None; + } +} + +fn file_nonblocking() -> Option { + if !FILE_ON.load(Ordering::Relaxed) { + return None; + } + let mut g = FILE_SINK.lock().unwrap_or_else(|e| e.into_inner()); + if g.is_none() { + if !FILE_ON.load(Ordering::Relaxed) { + return None; + } + let dir = log_dir(); + let _ = fs::create_dir_all(&dir); + let appender = tracing_appender::rolling::daily(&dir, "qmonitor.log"); + let (nb, guard) = tracing_appender::non_blocking(appender); + *g = Some((nb.clone(), guard)); + return Some(nb); + } + g.as_ref().map(|(nb, _)| nb.clone()) +} + +#[derive(Clone, Copy)] +struct GatedMakeWriter; + +struct GatedWriter { + inner: Option, +} + +impl Write for GatedWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + match self.inner.as_mut() { + Some(w) => w.write(buf), + None => Ok(buf.len()), + } + } + + fn flush(&mut self) -> io::Result<()> { + match self.inner.as_mut() { + Some(w) => w.flush(), + None => Ok(()), + } + } +} + +impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for GatedMakeWriter { + type Writer = GatedWriter; + + fn make_writer(&'a self) -> Self::Writer { + GatedWriter { + inner: file_nonblocking(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::Ordering; + + #[test] + fn loop_alive_false_without_detect() { + let h = RuntimeHealth::default(); + assert!(!h.loop_alive(3)); + } + + #[test] + fn loop_alive_true_when_recent() { + let h = RuntimeHealth { + last_detect_at: Some(Utc::now()), + ..Default::default() + }; + assert!(h.loop_alive(3)); + } + + #[test] + fn tooltip_includes_db_state() { + let h = RuntimeHealth { + last_detect_at: Some(Utc::now()), + db_ok: true, + ..Default::default() + }; + let tip = h.tray_tooltip(3); + assert!(tip.contains("DB ok"), "{tip}"); + assert!(tip.contains("qMonitor"), "{tip}"); + } + + #[test] + fn prune_deletes_oversized_logs_keeps_other_files() { + let dir = tempfile::tempdir().unwrap(); + let big = dir.path().join("qmonitor.log.2026-01-01"); + fs::write(&big, vec![b'y'; 200]).unwrap(); + let other = dir.path().join("notes.txt"); + fs::write(&other, b"keep").unwrap(); + + prune_log_dir(dir.path(), 3, 50); + assert!(!big.exists(), "size-pruned"); + assert!(other.exists(), "non-log kept"); + } + + #[test] + fn prune_zero_days_deletes_all_logs() { + let dir = tempfile::tempdir().unwrap(); + let log = dir.path().join("qmonitor.log"); + fs::write(&log, b"x").unwrap(); + prune_log_dir(dir.path(), 0, 0); + assert!(!log.exists()); + } + + #[test] + fn prune_deletes_old_logs() { + let dir = tempfile::tempdir().unwrap(); + let old = dir.path().join("qmonitor.log.old"); + fs::write(&old, vec![b'x'; 100]).unwrap(); + let old_time = SystemTime::now() - Duration::from_secs(10 * 86_400); + let ok = fs::OpenOptions::new() + .write(true) + .open(&old) + .and_then(|f| f.set_modified(old_time)) + .is_ok(); + if !ok { + return; + } + prune_log_dir(dir.path(), 3, LOG_MAX_BYTES); + assert!(!old.exists(), "age-pruned"); + } + + #[test] + fn prune_keeps_active_daily_log_over_size_cap() { + let dir = tempfile::tempdir().unwrap(); + FILE_ON.store(true, Ordering::Relaxed); + let appender = tracing_appender::rolling::daily(dir.path(), "qmonitor.log"); + let (nb, guard) = tracing_appender::non_blocking(appender); + *FILE_SINK.lock().unwrap_or_else(|e| e.into_inner()) = Some((nb, guard)); + struct ResetSink; + impl Drop for ResetSink { + fn drop(&mut self) { + drop_file_sink(); + FILE_ON.store(false, Ordering::Relaxed); + } + } + let _reset = ResetSink; + + let active = dir.path().join( + active_daily_log_name().expect("sink is active"), + ); + fs::write(&active, vec![b'x'; LOG_MAX_BYTES as usize + 64]).unwrap(); + let other = dir.path().join("qmonitor.log.2020-01-01"); + fs::write(&other, vec![b'y'; 200]).unwrap(); + + prune_log_dir(dir.path(), 3, LOG_MAX_BYTES); + assert!(active.exists(), "active daily log preserved"); + assert!(!other.exists(), "other logs size-pruned"); + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 852292e..f895530 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3,10 +3,14 @@ mod config; mod db; mod detect; mod device; +mod health; mod identity; +mod live_session; mod oauth_loopback; +mod persist; mod pkce; mod push; +mod runtime; mod session; mod update_check; @@ -60,6 +64,7 @@ async fn save_config( let channel_changed = prev_channel != config.update_channel; let next_channel = config.update_channel; config.save()?; + crate::health::apply_log_level(config.log_level); *state.config.write().await = config.clone(); state.reload_pipeline().await; if url_changed { @@ -239,11 +244,25 @@ async fn add_manual_game( #[tauri::command] async fn open_db(state: State<'_, Arc>) -> Result { - state.connect_db().await?; let path = state.config.read().await.resolved_db_path(); + let dir = path + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .ok_or_else(|| "invalid database path".to_string())?; + std::fs::create_dir_all(dir).map_err(|e| e.to_string())?; + // Open the folder, not the .db file — there is usually no default app. + open::that(dir).map_err(|e| e.to_string())?; Ok(path.display().to_string()) } +#[tauri::command] +fn open_log_dir() -> Result { + let dir = crate::health::log_dir(); + let _ = std::fs::create_dir_all(&dir); + open::that(&dir).map_err(|e| e.to_string())?; + Ok(dir.display().to_string()) +} + #[tauri::command] async fn is_onboarded(state: State<'_, Arc>) -> Result { let cfg = state.config.read().await.clone(); @@ -295,9 +314,7 @@ fn emit_update_event(app: &tauri::AppHandle, pending: Option<&PendingUpdate>) { #[cfg_attr(mobile, tauri::mobile_entry_point)] pub fn run() { - tracing_subscriber::fmt() - .with_env_filter("qmonitor=info,warn") - .init(); + crate::health::init_tracing(); let app_state = Arc::new(AppState::new()); let login_listener = LoginListener(Arc::new(Mutex::new(None))); @@ -306,6 +323,7 @@ pub fn run() { tauri::Builder::default() .plugin(tauri_plugin_opener::init()) .plugin(tauri_plugin_shell::init()) + .plugin(tauri_plugin_dialog::init()) .plugin(tauri_plugin_autostart::init( MacosLauncher::LaunchAgent, Some(vec!["--autostart"]), @@ -350,20 +368,7 @@ pub fn run() { }) .build(app)?; - tauri::async_runtime::spawn(async move { - if let Err(e) = state.connect_db().await { - tracing::error!(%e, "initial local database connect failed; will retry on tick/home"); - } - loop { - let interval = { - let cfg = state.config.read().await; - cfg.poll_interval_secs.max(1) - }; - state.tick().await; - let _ = handle.emit("qmonitor://tick", ()); - tokio::time::sleep(std::time::Duration::from_secs(interval)).await; - } - }); + crate::runtime::spawn_workers(handle.clone(), state.clone(), _tray); let detectable_state = app_state.clone(); tauri::async_runtime::spawn(async move { @@ -440,6 +445,7 @@ pub fn run() { unignore_game, add_manual_game, open_db, + open_log_dir, is_onboarded, login_listening, get_app_version, diff --git a/src-tauri/src/live_session.rs b/src-tauri/src/live_session.rs new file mode 100644 index 0000000..3b8e553 --- /dev/null +++ b/src-tauri/src/live_session.rs @@ -0,0 +1,217 @@ +//! In-memory play tracker. Detect always updates this; persist is best-effort. + +use chrono::{DateTime, Duration, Utc}; + +use crate::identity::GameIdentity; + +/// Wall-clock grace before a missing primary counts as quit. +pub const MISS_GRACE: Duration = Duration::seconds(8); +/// Unobserved gap treated as sleep/hang — split the session. +pub const SLEEP_SPLIT: Duration = Duration::seconds(30); + +#[derive(Debug, Clone)] +pub struct DetectSample { + pub observed_at: DateTime, + pub primary: Option, +} + +impl DetectSample { + pub fn empty() -> Self { + Self { + observed_at: Utc::now(), + primary: None, + } + } +} + +#[derive(Debug, Clone)] +pub struct PendingEnd { + pub identity: GameIdentity, + pub db_session_id: Option, + pub started_at: DateTime, + pub ended_at: DateTime, +} + +#[derive(Debug, Clone, Default)] +pub struct LiveSession { + pub identity: Option, + pub started_at: Option>, + pub last_seen_at: Option>, + pub db_session_id: Option, + pub pending_ends: Vec, + pub last_tick_at: Option>, +} + +impl LiveSession { + pub fn identity_id(&self) -> Option<&str> { + self.identity.as_ref().map(|i| i.id.as_str()) + } + + /// Apply a detect sample. Never waits on I/O. + pub fn apply(&mut self, sample: &DetectSample) { + let now = sample.observed_at; + let gap = self + .last_tick_at + .map(|t| now.signed_duration_since(t)); + self.last_tick_at = Some(now); + + if gap.is_some_and(|g| g > SLEEP_SPLIT) && self.identity.is_some() { + self.queue_end(); + } + + match &sample.primary { + Some(primary) => { + if self.identity.as_ref().is_some_and(|cur| cur.id == primary.id) { + self.last_seen_at = Some(now); + self.identity = Some(primary.clone()); + } else { + if self.identity.is_some() { + self.queue_end(); + } + self.identity = Some(primary.clone()); + self.started_at = Some(now); + self.last_seen_at = Some(now); + self.db_session_id = None; + } + } + None => { + if self.identity.is_some() { + let last = self.last_seen_at.unwrap_or(now); + if now.signed_duration_since(last) >= MISS_GRACE { + self.queue_end(); + } + } + } + } + } + + pub fn clear_identity(&mut self) { + self.identity = None; + self.started_at = None; + self.last_seen_at = None; + self.db_session_id = None; + } + + fn queue_end(&mut self) { + let Some(identity) = self.identity.take() else { + return; + }; + let started_at = self + .started_at + .or(self.last_seen_at) + .unwrap_or_else(Utc::now); + let ended_at = self.last_seen_at.unwrap_or(started_at); + let db_session_id = self.db_session_id.take(); + self.started_at = None; + self.last_seen_at = None; + self.pending_ends.push(PendingEnd { + identity, + db_session_id, + started_at, + ended_at, + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::identity::{Confidence, GameIdentity}; + + fn game(id: &str, title: &str) -> GameIdentity { + GameIdentity { + id: id.into(), + title: title.into(), + steam_app_id: None, + exe: Some(format!("{title}.exe")), + confidence: Confidence::High, + source: "test".into(), + fingerprint: None, + } + } + + fn sample_at(at: DateTime, primary: Option) -> DetectSample { + DetectSample { + observed_at: at, + primary, + } + } + + #[test] + fn tracks_start_without_db() { + let mut live = LiveSession::default(); + let t0 = Utc::now(); + live.apply(&sample_at(t0, Some(game("steam:1", "RL")))); + assert_eq!(live.identity_id(), Some("steam:1")); + assert_eq!(live.started_at, Some(t0)); + assert!(live.pending_ends.is_empty()); + } + + #[test] + fn pending_end_after_miss_grace_even_without_db() { + let mut live = LiveSession::default(); + let t0 = Utc::now(); + live.apply(&sample_at(t0, Some(game("steam:1", "RL")))); + live.apply(&sample_at(t0 + Duration::seconds(3), None)); + assert!(live.identity.is_some(), "still in grace"); + assert!(live.pending_ends.is_empty()); + + live.apply(&sample_at(t0 + Duration::seconds(9), None)); + assert!(live.identity.is_none()); + assert_eq!(live.pending_ends.len(), 1); + assert_eq!(live.pending_ends[0].ended_at, t0); + assert_eq!(live.pending_ends[0].started_at, t0); + } + + #[test] + fn hang_with_game_gone_ends_immediately() { + let mut live = LiveSession::default(); + let t0 = Utc::now(); + live.apply(&sample_at(t0, Some(game("steam:1", "RL")))); + live.apply(&sample_at(t0 + Duration::hours(1), None)); + assert_eq!(live.pending_ends.len(), 1); + assert_eq!(live.pending_ends[0].ended_at, t0); + assert!(live.identity.is_none()); + } + + #[test] + fn sleep_split_while_still_playing() { + let mut live = LiveSession::default(); + let t0 = Utc::now(); + let rl = game("steam:1", "RL"); + live.apply(&sample_at(t0, Some(rl.clone()))); + live.db_session_id = Some("db-1".into()); + let t1 = t0 + Duration::hours(2); + live.apply(&sample_at(t1, Some(rl))); + assert_eq!(live.pending_ends.len(), 1); + assert_eq!(live.pending_ends[0].db_session_id.as_deref(), Some("db-1")); + assert_eq!(live.pending_ends[0].ended_at, t0); + assert_eq!(live.identity_id(), Some("steam:1")); + assert_eq!(live.started_at, Some(t1)); + assert!(live.db_session_id.is_none()); + } + + #[test] + fn identity_switch_queues_end_then_starts() { + let mut live = LiveSession::default(); + let t0 = Utc::now(); + live.apply(&sample_at(t0, Some(game("steam:1", "A")))); + live.apply(&sample_at(t0 + Duration::seconds(3), Some(game("steam:2", "B")))); + assert_eq!(live.pending_ends.len(), 1); + assert_eq!(live.pending_ends[0].identity.id, "steam:1"); + assert_eq!(live.identity_id(), Some("steam:2")); + } + + #[test] + fn flicker_within_grace_does_not_end() { + let mut live = LiveSession::default(); + let t0 = Utc::now(); + let rl = game("steam:1", "RL"); + live.apply(&sample_at(t0, Some(rl.clone()))); + live.apply(&sample_at(t0 + Duration::seconds(3), None)); + live.apply(&sample_at(t0 + Duration::seconds(6), Some(rl))); + assert!(live.pending_ends.is_empty()); + assert_eq!(live.started_at, Some(t0)); + assert_eq!(live.last_seen_at, Some(t0 + Duration::seconds(6))); + } +} diff --git a/src-tauri/src/persist.rs b/src-tauri/src/persist.rs new file mode 100644 index 0000000..62ad006 --- /dev/null +++ b/src-tauri/src/persist.rs @@ -0,0 +1,585 @@ +//! Persist worker: exclusive owner of the Turso connection. + +use std::future::Future; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use tokio::sync::{mpsc, watch}; +use tokio::time::timeout; + +use crate::db::{SessionRow, TursoDb}; +use crate::identity::{GameIdentity, ManualGame}; +use crate::live_session::{DetectSample, PendingEnd, SLEEP_SPLIT}; +#[cfg(test)] +use crate::live_session::LiveSession; +use crate::session::AppState; + +pub const DB_OP_TIMEOUT: Duration = Duration::from_secs(2); +pub const DB_OPEN_TIMEOUT: Duration = Duration::from_secs(5); +pub const RECONNECT_BACKOFF: Duration = Duration::from_secs(5); + +#[derive(Debug)] +pub enum PersistError { + Poison(String), + PathChanged, +} + +pub enum PersistCmd { + Confirm { + fingerprint: String, + title: String, + reply: tokio::sync::oneshot::Sender>, + }, + Ignore { + identity_id: String, + title: String, + reply: tokio::sync::oneshot::Sender>, + }, + Unignore { + identity_id: String, + reply: tokio::sync::oneshot::Sender>, + }, + AddManual { + game: ManualGame, + identity_id: String, + reply: tokio::sync::oneshot::Sender>, + }, + EnsureOpen { + reply: tokio::sync::oneshot::Sender>, + }, +} + +#[derive(Debug, Clone, Default)] +pub struct DbView { + pub turso_ok: bool, + pub pending_count: i64, + pub active: Option, + pub history: Vec, +} + +pub async fn timed(fut: impl Future>) -> Result { + match timeout(DB_OP_TIMEOUT, fut).await { + Ok(Ok(v)) => Ok(v), + Ok(Err(e)) => Err(PersistError::Poison(e)), + Err(_) => Err(PersistError::Poison("db op timeout".into())), + } +} + +fn persist_reply(res: &Result<(), PersistError>) -> Result<(), String> { + match res { + Ok(()) => Ok(()), + Err(PersistError::Poison(e)) => Err(e.clone()), + Err(PersistError::PathChanged) => Err("db path changed".into()), + } +} + +/// Flush pending ends + current tracking onto the DB. DB start wins when a row exists. +#[cfg(test)] +pub async fn flush_live(db: &TursoDb, live: &mut LiveSession) -> Result<(), String> { + let pending = std::mem::take(&mut live.pending_ends); + for end in pending { + write_pending_end(db, &end).await?; + } + if let Some((id, started)) = reconcile_live( + db, + live.identity.as_ref(), + live.started_at, + live.last_seen_at, + live.last_tick_at, + ) + .await? + { + live.db_session_id = Some(id); + live.started_at = Some(started); + } + Ok(()) +} + +async fn write_pending_end(db: &TursoDb, end: &PendingEnd) -> Result<(), String> { + if let Some(id) = &end.db_session_id { + if let Some(row) = db.get_session(id).await? { + if row.push_status == crate::db::PushStatus::Active { + db.end_session_at(id, end.ended_at).await?; + return Ok(()); + } + } + } + let actives = db.list_active().await?; + let mut found = false; + for row in actives.iter().filter(|s| s.identity_id == end.identity.id) { + db.end_session_at(&row.id, end.ended_at).await?; + found = true; + } + if !found { + let opened = db + .open_session_at(&end.identity, end.started_at) + .await?; + db.end_session_at(&opened.id, end.ended_at).await?; + } + Ok(()) +} + +/// Open-or-reuse the tracked identity, end other actives, discard duplicate actives, +/// and cap orphan endings at `SLEEP_SPLIT`. +async fn reconcile_live( + db: &TursoDb, + identity: Option<&GameIdentity>, + started_at: Option>, + last_seen_at: Option>, + last_tick_at: Option>, +) -> Result)>, String> { + if let Some(identity) = identity { + let started = started_at.unwrap_or_else(Utc::now); + let row = db.open_session_at(identity, started).await?; + let mut keep_id = row.id.clone(); + let mut keep_started = row.started_at; + + let actives = db.list_active().await?; + let last_seen = last_seen_at.unwrap_or(row.started_at); + for other in actives.iter().filter(|s| s.identity_id != identity.id) { + db.end_session_at(&other.id, last_seen).await?; + } + + let mut same: Vec<_> = db + .list_active() + .await? + .into_iter() + .filter(|s| s.identity_id == identity.id) + .collect(); + same.sort_by_key(|s| s.started_at); + if same.len() > 1 { + let discard: Vec = same.iter().skip(1).map(|s| s.id.clone()).collect(); + db.discard_active_sessions(&discard).await?; + } + if let Some(keep) = same.first() { + keep_id = keep.id.clone(); + keep_started = keep.started_at; + } + Ok(Some((keep_id, keep_started))) + } else if last_tick_at.is_some() { + for row in db.list_active().await? { + let cap = row.started_at + SLEEP_SPLIT; + let ended = Utc::now().min(cap); + db.end_session_at(&row.id, ended).await?; + } + Ok(None) + } else { + Ok(None) + } +} + +async fn refresh_view(db: &TursoDb) -> Result { + Ok(DbView { + turso_ok: true, + pending_count: db.count_pending().await?, + active: db.get_active().await?, + history: db.list_sessions(100).await?, + }) +} + +async fn load_prefs(db: &TursoDb, state: &AppState) -> Result<(), String> { + let mappings = db.list_mappings().await?; + let ignored = db.list_ignored().await?; + let manuals = db.list_manual_games().await?; + { + let mut titles = state.ignored_titles.write().await; + *titles = ignored + .iter() + .map(|i| (i.identity_id.clone(), i.title.clone())) + .collect(); + } + let mut pipe = state.pipeline.write().await; + pipe.user_mappings = mappings + .into_iter() + .map(|m| (m.fingerprint.clone(), m)) + .collect(); + pipe.ignored_identities = ignored.into_iter().map(|i| i.identity_id).collect(); + pipe.manual_games = manuals; + Ok(()) +} + +pub async fn run_persist( + state: Arc, + mut sample_rx: watch::Receiver, + mut cmd_rx: mpsc::Receiver, + push_tx: mpsc::Sender, + mut push_result_rx: mpsc::Receiver<(String, Result<(), String>)>, +) { + let mut generation: u64 = 0; + loop { + let path = state.config.read().await.resolved_db_path(); + generation += 1; + if generation > 1 { + let mut h = state.health.write().await; + h.db_reconnects += 1; + h.db_ok = false; + h.db_generation = generation; + } + match open_db(&path).await { + Ok(db) => { + tracing::info!(gen = generation, path = %path.display(), "persist db open"); + { + let mut h = state.health.write().await; + h.db_ok = true; + h.db_generation = generation; + h.last_persist_at = Some(Utc::now()); + } + if let Err(e) = load_prefs(&db, &state).await { + tracing::warn!(%e, "load pipeline prefs failed"); + } + match persist_loop( + &state, + &db, + &path, + &mut sample_rx, + &mut cmd_rx, + &push_tx, + &mut push_result_rx, + ) + .await + { + Ok(()) => return, + Err(PersistError::PathChanged) => { + tracing::info!(gen = generation, "db path changed; reopening"); + } + Err(PersistError::Poison(e)) => { + tracing::warn!(%e, gen = generation, "persist poisoned; dropping connection"); + *state.last_error.write().await = Some(e.clone()); + state.health.write().await.db_ok = false; + state.health.write().await.last_error = Some(e); + *state.db_view.write().await = DbView::default(); + } + } + drop(db); + } + Err(e) => { + tracing::error!(%e, "persist db open failed"); + *state.last_error.write().await = Some(e.clone()); + state.health.write().await.db_ok = false; + state.health.write().await.last_error = Some(e); + *state.db_view.write().await = DbView::default(); + } + } + tokio::time::sleep(RECONNECT_BACKOFF).await; + } +} + +async fn open_db(path: &PathBuf) -> Result { + timeout(DB_OPEN_TIMEOUT, TursoDb::open(path)) + .await + .map_err(|_| "db open timeout".to_string())? +} + +async fn persist_loop( + state: &AppState, + db: &TursoDb, + opened_path: &Path, + sample_rx: &mut watch::Receiver, + cmd_rx: &mut mpsc::Receiver, + push_tx: &mpsc::Sender, + push_result_rx: &mut mpsc::Receiver<(String, Result<(), String>)>, +) -> Result<(), PersistError> { + let mut purge_ticks: u32 = 0; + loop { + tokio::select! { + biased; + r = sample_rx.changed() => { + if r.is_err() { + return Ok(()); + } + apply_sample(state, db, push_tx).await?; + } + Some(cmd) = cmd_rx.recv() => { + handle_cmd(state, db, opened_path, cmd).await?; + } + Some((id, result)) = push_result_rx.recv() => { + match result { + Ok(()) => timed(db.mark_synced(&id)).await?, + Err(e) => { + let retry = match timed(db.get_session(&id)).await { + Ok(Some(r)) => r.retry_count + 1, + _ => 1, + }; + let _ = timed(db.mark_push_failed(&id, &e, retry)).await; + *state.last_error.write().await = Some(e.clone()); + } + } + refresh_and_store(state, db).await?; + } + _ = tokio::time::sleep(Duration::from_secs(2)) => { + apply_sample(state, db, push_tx).await?; + purge_ticks += 1; + if purge_ticks >= 30 { + purge_ticks = 0; + let days = state.config.read().await.retention_acked_days; + let _ = timed(db.purge_synced(days)).await; + } + } + } + } +} + +async fn apply_sample( + state: &AppState, + db: &TursoDb, + push_tx: &mpsc::Sender, +) -> Result<(), PersistError> { + let pending = { + let mut live = state.live.write().await; + std::mem::take(&mut live.pending_ends) + }; + let mut pending = pending.into_iter(); + while let Some(end) = pending.next() { + if let Err(e) = timed(write_pending_end(db, &end)).await { + let mut live = state.live.write().await; + let mut rest: Vec<_> = std::iter::once(end).chain(pending).collect(); + rest.append(&mut live.pending_ends); + live.pending_ends = rest; + return Err(e); + } + } + + let snapshot = state.live.read().await.clone(); + let keep = timed(reconcile_live( + db, + snapshot.identity.as_ref(), + snapshot.started_at, + snapshot.last_seen_at, + snapshot.last_tick_at, + )) + .await?; + if let (Some(identity), Some((id, started))) = (snapshot.identity.as_ref(), keep) { + let mut live = state.live.write().await; + if live.identity_id() == Some(identity.id.as_str()) { + live.db_session_id = Some(id); + live.started_at = Some(started); + } + } + + enqueue_due(db, push_tx).await?; + refresh_and_store(state, db).await?; + state.health.write().await.last_persist_at = Some(Utc::now()); + Ok(()) +} + +async fn enqueue_due(db: &TursoDb, push_tx: &mpsc::Sender) -> Result<(), PersistError> { + let due = timed(db.list_due_pushes()).await?; + for row in due { + if push_tx.try_send(row).is_err() { + break; + } + } + Ok(()) +} + +async fn refresh_and_store(state: &AppState, db: &TursoDb) -> Result<(), PersistError> { + let view = timed(refresh_view(db)).await?; + *state.db_view.write().await = view; + Ok(()) +} + +async fn handle_cmd( + state: &AppState, + db: &TursoDb, + opened_path: &Path, + cmd: PersistCmd, +) -> Result<(), PersistError> { + match cmd { + PersistCmd::Confirm { + fingerprint, + title, + reply, + } => { + let identity_id = format!("user:{fingerprint}"); + let res = timed(async { + db.upsert_mapping(&fingerprint, &title, &identity_id) + .await?; + let _ = db.remove_ignored(&identity_id).await; + Ok(()) + }) + .await; + let out = persist_reply(&res); + let _ = reply.send(out); + res?; + load_prefs(db, state).await.ok(); + } + PersistCmd::Ignore { + identity_id, + title, + reply, + } => { + let id = identity_id.clone(); + let res = timed(async { + db.upsert_ignored(&id, &title).await?; + let actives = db.list_active().await.unwrap_or_default(); + let discard: Vec = actives + .into_iter() + .filter(|s| s.identity_id == id) + .map(|s| s.id) + .collect(); + if !discard.is_empty() { + db.discard_active_sessions(&discard).await?; + } + Ok(()) + }) + .await; + let out = persist_reply(&res); + let _ = reply.send(out); + res?; + { + let mut live = state.live.write().await; + if live.identity_id() == Some(identity_id.as_str()) { + live.clear_identity(); + } + } + load_prefs(db, state).await.ok(); + } + PersistCmd::Unignore { + identity_id, + reply, + } => { + let res = timed(db.remove_ignored(&identity_id)).await; + let out = persist_reply(&res); + let _ = reply.send(out); + res?; + load_prefs(db, state).await.ok(); + } + PersistCmd::AddManual { + game, + identity_id, + reply, + } => { + let res = timed(async { + db.upsert_manual_game(&game).await?; + let _ = db.remove_ignored(&identity_id).await; + Ok(()) + }) + .await; + let out = persist_reply(&res); + let _ = reply.send(out); + res?; + load_prefs(db, state).await.ok(); + } + PersistCmd::EnsureOpen { reply } => { + let configured = state.config.read().await.resolved_db_path(); + if configured.as_path() != opened_path { + let _ = reply.send(Ok(configured.display().to_string())); + return Err(PersistError::PathChanged); + } + match timed(db.ping()).await { + Ok(()) => { + let _ = reply.send(Ok(configured.display().to_string())); + } + Err(PersistError::Poison(e)) => { + let _ = reply.send(Err(e.clone())); + return Err(PersistError::Poison(e)); + } + Err(PersistError::PathChanged) => return Err(PersistError::PathChanged), + } + } + } + refresh_and_store(state, db).await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::identity::{Confidence, GameIdentity}; + use chrono::Duration as ChronoDuration; + use tempfile::tempdir; + + fn identity(id: &str, title: &str) -> GameIdentity { + GameIdentity { + id: id.into(), + title: title.into(), + steam_app_id: None, + exe: Some(format!("{title}.exe")), + confidence: Confidence::High, + source: "test".into(), + fingerprint: None, + } + } + + #[tokio::test] + async fn timed_poison_on_timeout() { + let err = timed(async { + tokio::time::sleep(Duration::from_secs(30)).await; + Ok::<(), String>(()) + }) + .await + .unwrap_err(); + match err { + PersistError::Poison(msg) => assert!(msg.contains("timeout"), "{msg}"), + PersistError::PathChanged => panic!("expected timeout poison"), + } + } + + #[tokio::test] + async fn recover_end_orphans_instead_of_discard() { + let dir = tempdir().unwrap(); + let db = TursoDb::open(dir.path().join("rec.db")).await.unwrap(); + let apex = identity("steam:1172470", "Apex"); + let row = db + .force_insert_active(&apex, Utc::now() - ChronoDuration::hours(1)) + .await + .unwrap(); + let mut live = LiveSession { + last_tick_at: Some(Utc::now()), + ..Default::default() + }; + flush_live(&db, &mut live).await.unwrap(); + assert!(db.list_active().await.unwrap().is_empty()); + let due = db.list_due_pushes().await.unwrap(); + assert_eq!(due.len(), 1); + assert_eq!(due[0].id, row.id); + assert!(db.get_session(&row.id).await.unwrap().is_some()); + } + + #[tokio::test] + async fn pending_end_without_db_row_inserts_and_ends() { + let dir = tempdir().unwrap(); + let db = TursoDb::open(dir.path().join("mem.db")).await.unwrap(); + let apex = identity("steam:1", "RL"); + let t0 = Utc::now() - ChronoDuration::hours(1); + let t1 = t0 + ChronoDuration::minutes(50); + let mut live = LiveSession { + pending_ends: vec![PendingEnd { + identity: apex, + db_session_id: None, + started_at: t0, + ended_at: t1, + }], + last_tick_at: Some(Utc::now()), + ..Default::default() + }; + flush_live(&db, &mut live).await.unwrap(); + assert!(db.list_active().await.unwrap().is_empty()); + let due = db.list_due_pushes().await.unwrap(); + assert_eq!(due.len(), 1); + assert_eq!(due[0].started_at, t0); + assert_eq!(due[0].ended_at, Some(t1)); + } + + #[tokio::test] + async fn db_active_started_at_wins_over_memory() { + let dir = tempdir().unwrap(); + let db = TursoDb::open(dir.path().join("win.db")).await.unwrap(); + let apex = identity("steam:1", "RL"); + let db_start = Utc::now() - ChronoDuration::minutes(20); + let opened = db.force_insert_active(&apex, db_start).await.unwrap(); + let mut live = LiveSession { + identity: Some(apex), + started_at: Some(Utc::now()), + last_seen_at: Some(Utc::now()), + last_tick_at: Some(Utc::now()), + ..Default::default() + }; + flush_live(&db, &mut live).await.unwrap(); + assert_eq!(live.db_session_id.as_deref(), Some(opened.id.as_str())); + assert_eq!(live.started_at, Some(db_start)); + assert_eq!(db.list_active().await.unwrap().len(), 1); + } +} diff --git a/src-tauri/src/push.rs b/src-tauri/src/push.rs index f530961..4720768 100644 --- a/src-tauri/src/push.rs +++ b/src-tauri/src/push.rs @@ -1,11 +1,17 @@ //! Webhook push client for completed sessions. +use std::time::Duration; + use serde::Serialize; use crate::auth; use crate::config::AppConfig; use crate::db::SessionRow; +pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(8); +pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(15); +pub const POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(60); + #[derive(Debug, Serialize)] pub struct SessionPayload { pub schema_version: u32, @@ -51,19 +57,30 @@ impl SessionPayload { } } +pub fn http_client() -> reqwest::Client { + reqwest::Client::builder() + .connect_timeout(CONNECT_TIMEOUT) + .timeout(REQUEST_TIMEOUT) + .pool_idle_timeout(POOL_IDLE_TIMEOUT) + .build() + .expect("reqwest client") +} + pub struct WebhookClient { http: reqwest::Client, } impl Default for WebhookClient { fn default() -> Self { - Self { - http: reqwest::Client::new(), - } + Self::new() } } impl WebhookClient { + pub fn new() -> Self { + Self { http: http_client() } + } + pub async fn push( &self, cfg: &AppConfig, @@ -134,4 +151,13 @@ mod tests { assert_eq!(p.steam_app_id, Some(570)); assert_eq!(p.duration_secs, 120); } + + #[test] + fn http_client_builds_with_timeouts() { + assert_eq!(CONNECT_TIMEOUT, Duration::from_secs(8)); + assert_eq!(REQUEST_TIMEOUT, Duration::from_secs(15)); + assert_eq!(POOL_IDLE_TIMEOUT, Duration::from_secs(60)); + let _ = http_client(); + let _ = WebhookClient::new(); + } } diff --git a/src-tauri/src/runtime.rs b/src-tauri/src/runtime.rs new file mode 100644 index 0000000..92776f0 --- /dev/null +++ b/src-tauri/src/runtime.rs @@ -0,0 +1,175 @@ +//! Isolated detect / persist / push / health workers. + +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use tauri::Emitter; +use tokio::sync::{mpsc, watch}; +use tokio::task::JoinHandle; + +use crate::auth; +use crate::db::SessionRow; +use crate::detect::{foreground_pid, primary_identity, snapshot_processes}; +use crate::health::RuntimeHealth; +use crate::identity::ProcessSnapshot; +use crate::live_session::DetectSample; +use crate::persist::{self, PersistCmd}; +use crate::push::WebhookClient; +use crate::session::AppState; + +const DETECT_BLOCKING_TIMEOUT: Duration = Duration::from_secs(3); +const PUSH_POLL: Duration = Duration::from_secs(2); +const HEALTH_PULSE: Duration = Duration::from_secs(3); + +pub fn spawn_workers(app: tauri::AppHandle, state: Arc, tray: tauri::tray::TrayIcon) { + let (sample_tx, sample_rx) = watch::channel(DetectSample::empty()); + let (cmd_tx, cmd_rx) = mpsc::channel::(32); + let (push_tx, push_rx) = mpsc::channel::(1); + let (push_result_tx, push_result_rx) = mpsc::channel::<(String, Result<(), String>)>(8); + + { + let mut slot = state.persist_tx.lock().expect("persist_tx"); + *slot = Some(cmd_tx); + } + + let detect_state = state.clone(); + tauri::async_runtime::spawn(async move { + run_detect(detect_state, sample_tx).await; + }); + + let persist_state = state.clone(); + tauri::async_runtime::spawn(async move { + persist::run_persist(persist_state, sample_rx, cmd_rx, push_tx, push_result_rx).await; + }); + + let push_state = state.clone(); + tauri::async_runtime::spawn(async move { + run_push(push_state, push_rx, push_result_tx).await; + }); + + let health_state = state.clone(); + tauri::async_runtime::spawn(async move { + run_health(app, health_state, tray).await; + }); +} + +async fn run_detect(state: Arc, sample_tx: watch::Sender) { + let mut prev_processes: Vec = Vec::new(); + let mut prev_fg: Option = None; + let mut in_flight: Option, Option)>> = None; + loop { + let interval = state.config.read().await.poll_interval_secs.max(1); + if in_flight.is_none() { + in_flight = Some(tokio::task::spawn_blocking(|| { + let processes = snapshot_processes(); + let fg = foreground_pid(); + (processes, fg) + })); + } + + let snap = tokio::time::timeout( + DETECT_BLOCKING_TIMEOUT, + in_flight.as_mut().expect("in-flight snapshot"), + ) + .await; + + let (processes, fg) = match snap { + Ok(Ok(pair)) => { + in_flight = None; + prev_processes = pair.0.clone(); + prev_fg = pair.1; + pair + } + Ok(Err(e)) => { + in_flight = None; + tracing::warn!(%e, "detect join failed"); + (prev_processes.clone(), prev_fg) + } + Err(_) => { + state.health.write().await.detect_timeouts += 1; + tracing::warn!("detect snapshot timed out; keeping previous sample"); + (prev_processes.clone(), prev_fg) + } + }; + + let (identities, pending) = { + let pipe = state.pipeline.read().await; + pipe.resolve_running(&processes) + }; + *state.pending_detections.write().await = pending; + let primary = primary_identity(&identities, fg, &processes).cloned(); + + if let Some(p) = &primary { + let pid = processes + .iter() + .find(|proc| { + p.exe + .as_ref() + .map(|e| e.eq_ignore_ascii_case(&proc.name)) + .unwrap_or(false) + }) + .map(|proc| proc.pid); + tracing::debug!( + id = %p.id, + title = %p.title, + ?pid, + "detect primary" + ); + } + + let sample = DetectSample { + observed_at: Utc::now(), + primary, + }; + state.live.write().await.apply(&sample); + state.health.write().await.last_detect_at = Some(sample.observed_at); + let _ = sample_tx.send(sample); + + tokio::time::sleep(Duration::from_secs(interval)).await; + } +} + +async fn run_push( + state: Arc, + mut push_rx: mpsc::Receiver, + result_tx: mpsc::Sender<(String, Result<(), String>)>, +) { + let client = WebhookClient::new(); + loop { + let cfg = state.config.read().await.clone(); + let (Some(url), Some(token)) = (cfg.webhook_url(), auth::get_access_token(&cfg)) else { + tokio::time::sleep(PUSH_POLL).await; + continue; + }; + let Some(row) = push_rx.recv().await else { + break; + }; + let result = client.push(&cfg, &url, &token, &row).await; + if result.is_ok() { + state.health.write().await.last_push_at = Some(Utc::now()); + } else if let Err(e) = &result { + *state.last_error.write().await = Some(e.clone()); + } + let _ = result_tx.send((row.id, result)).await; + } +} + +async fn run_health(app: tauri::AppHandle, state: Arc, tray: tauri::tray::TrayIcon) { + let mut last_prune = tokio::time::Instant::now(); + loop { + let (interval, log_level) = { + let cfg = state.config.read().await; + (cfg.poll_interval_secs.max(1), cfg.log_level) + }; + let health: RuntimeHealth = state.health.read().await.clone(); + let tip = health.tray_tooltip(interval); + let _ = tray.set_tooltip(Some(&tip)); + let _ = app.emit("qmonitor://tick", ()); + if last_prune.elapsed() >= crate::health::LOG_PRUNE_EVERY { + crate::health::prune_now(log_level); + last_prune = tokio::time::Instant::now(); + } + tokio::time::sleep(HEALTH_PULSE).await; + } +} diff --git a/src-tauri/src/session.rs b/src-tauri/src/session.rs index 5b720a8..d946705 100644 --- a/src-tauri/src/session.rs +++ b/src-tauri/src/session.rs @@ -1,21 +1,23 @@ -//! Session state machine: open on identity appear, end on disappear, push pending. +//! App state, pipeline prefs, and UI-facing home snapshot. -use std::collections::HashSet; +use std::collections::HashMap; +use std::future::Future; use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; +use std::time::Duration; use serde::Serialize; -use tokio::sync::RwLock; +use tokio::sync::{oneshot, RwLock}; use crate::auth; use crate::config::AppConfig; -use crate::db::{SessionRow, TursoDb}; -use crate::detect::{foreground_pid, primary_identity, snapshot_processes}; +use crate::db::{PushStatus, SessionRow, TursoDb}; +use crate::health::RuntimeHealth; use crate::identity::detectable::{self, DetectableCatalog, DETECTABLE_MAX_AGE}; use crate::identity::resolver::{parse_exe_input, IdentityPipeline, UserMapping}; use crate::identity::{ManualGame, PendingDetection, TrackableGame}; -use crate::identity::GameIdentity; -use crate::push::WebhookClient; +use crate::live_session::LiveSession; +use crate::persist::{DbView, PersistCmd}; #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -25,6 +27,10 @@ pub struct SyncStatus { pub last_error: Option, pub active_title: Option, pub webhook_configured: bool, + pub last_tick_at: Option, + pub loop_alive: bool, + pub db_reconnects: u64, + pub detect_timeouts: u64, } #[derive(Debug, Clone, Serialize)] @@ -38,12 +44,16 @@ pub struct HomeState { pub struct AppState { pub config: RwLock, + /// Test-only injected DB. Production persist owns the connection. pub db: RwLock>>, pub pipeline: RwLock, - pub active_identity_id: RwLock>, + pub live: RwLock, + pub db_view: RwLock, + pub health: RwLock, + pub persist_tx: Mutex>>, + pub ignored_titles: RwLock>, pub pending_detections: RwLock>, pub last_error: RwLock>, - pub webhook: WebhookClient, } impl AppState { @@ -74,113 +84,71 @@ impl AppState { config: RwLock::new(config), db: RwLock::new(None), pipeline: RwLock::new(pipeline), - active_identity_id: RwLock::new(None), + live: RwLock::new(LiveSession::default()), + db_view: RwLock::new(DbView::default()), + health: RwLock::new(RuntimeHealth::default()), + persist_tx: Mutex::new(None), + ignored_titles: RwLock::new(HashMap::new()), pending_detections: RwLock::new(Vec::new()), last_error: RwLock::new(None), - webhook: WebhookClient::default(), } } pub async fn connect_db(&self) -> Result<(), String> { - const ATTEMPTS: u32 = 8; - let mut last_err = String::from("db connect failed"); - for attempt in 0..ATTEMPTS { - match self.connect_db_once().await { - Ok(()) => { - // Clear a prior connect failure once we're healthy again. - let mut err = self.last_error.write().await; - if err - .as_ref() - .is_some_and(|e| e.starts_with("db connect:") || e.starts_with("db open:")) - { - *err = None; - } - return Ok(()); - } - Err(e) => { - last_err = e; - if attempt + 1 < ATTEMPTS { - let backoff_ms = 40u64.saturating_mul(2u64.pow(attempt.min(4))); - tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await; - } - } - } - } - let msg = format!("db connect: {last_err}"); - tracing::error!(%msg, attempts = ATTEMPTS, "local database unavailable"); - *self.last_error.write().await = Some(msg.clone()); - Err(msg) - } - - async fn connect_db_once(&self) -> Result<(), String> { - let cfg = self.config.read().await.clone(); - let path = cfg.resolved_db_path(); - let db = TursoDb::open(&path).await.map_err(|e| format!("db open: {e}"))?; - db.ping().await.map_err(|e| format!("db ping: {e}"))?; - self.load_pipeline_prefs(&db).await; - *self.db.write().await = Some(Arc::new(db)); - Ok(()) + self.call_persist(|reply| PersistCmd::EnsureOpen { reply }) + .await + .map(|_| ()) } - /// Open (or reopen) the local DB if missing or unresponsive. - pub async fn ensure_db(&self) -> Result, String> { - if let Some(db) = self.db.read().await.clone() { - if db.ping().await.is_ok() { - return Ok(db); - } - tracing::warn!("local database ping failed; reconnecting"); - *self.db.write().await = None; - } - self.connect_db().await?; - self.db - .read() - .await + async fn call_persist( + &self, + make: impl FnOnce(oneshot::Sender>) -> PersistCmd, + ) -> Result { + let tx = self + .persist_tx + .lock() + .map_err(|e| e.to_string())? .clone() - .ok_or_else(|| "db connect succeeded but handle missing".into()) + .ok_or_else(|| "persist offline".to_string())?; + let (reply, rx) = oneshot::channel(); + tx.send(make(reply)) + .await + .map_err(|_| "persist offline".to_string())?; + tokio::time::timeout(Duration::from_secs(8), rx) + .await + .map_err(|_| "persist timeout".to_string())? + .map_err(|_| "persist dropped".to_string())? } - async fn load_pipeline_prefs(&self, db: &TursoDb) { - let mappings = db.list_mappings().await.unwrap_or_default(); - let ignored = db.list_ignored().await.unwrap_or_default(); - let manuals = db.list_manual_games().await.unwrap_or_default(); - let mut pipe = self.pipeline.write().await; - pipe.user_mappings = mappings - .into_iter() - .map(|m| (m.fingerprint.clone(), m)) - .collect(); - pipe.ignored_identities = ignored.into_iter().map(|i| i.identity_id).collect(); - pipe.manual_games = manuals; + async fn persist_or_db( + &self, + make: impl FnOnce(oneshot::Sender>) -> PersistCmd, + fallback: F, + ) -> Result<(), String> + where + F: FnOnce() -> Fut, + Fut: Future>, + { + if self.persist_tx.lock().ok().and_then(|g| g.clone()).is_some() { + drop(fallback); + self.call_persist(make).await + } else { + fallback().await + } } pub async fn reload_pipeline(&self) { let cfg = self.config.read().await.clone(); let steam = cfg.steam_path_override.as_ref().map(PathBuf::from); let catalog = cfg.catalog_path.as_ref().map(PathBuf::from); - let (mappings, ignored, manuals): ( - std::collections::HashMap, - HashSet, - Vec, - ) = if let Some(db) = self.db.read().await.as_ref() { - let mappings = db - .list_mappings() - .await - .unwrap_or_default() - .into_iter() - .map(|m| (m.fingerprint.clone(), m)) - .collect(); - let ignored = db - .list_ignored() - .await - .unwrap_or_default() - .into_iter() - .map(|i| i.identity_id) - .collect(); - let manuals = db.list_manual_games().await.unwrap_or_default(); - (mappings, ignored, manuals) - } else { - (Default::default(), HashSet::new(), Vec::new()) + let (mappings, ignored, manuals) = { + let pipe = self.pipeline.read().await; + ( + pipe.user_mappings.clone(), + pipe.ignored_identities.clone(), + pipe.manual_games.clone(), + ) }; - // Preserve in-memory detectable if already loaded; otherwise load disk cache. let detectable = { let pipe = self.pipeline.read().await; if pipe.detectable.is_empty() { @@ -216,98 +184,33 @@ impl AppState { self.pipeline.write().await.detectable = catalog; } - pub async fn tick(&self) { - let processes = snapshot_processes(); - let (identities, pending) = { - let pipe = self.pipeline.read().await; - pipe.resolve_running(&processes) - }; - *self.pending_detections.write().await = pending; - - let fg = foreground_pid(); - let primary = primary_identity(&identities, fg, &processes).cloned(); - - let db = match self.ensure_db().await { - Ok(db) => db, - Err(e) => { - *self.last_error.write().await = Some(e); - return; - } - }; - - let prev = self.active_identity_id.read().await.clone(); - if let Err(e) = reconcile_active_sessions(&db, primary.as_ref(), prev.as_deref()).await { - *self.last_error.write().await = Some(e); - } - *self.active_identity_id.write().await = primary.as_ref().map(|i| i.id.clone()); - - // Push due sessions - self.flush_pushes(&db).await; - - // Periodic purge - let days = self.config.read().await.retention_acked_days; - let _ = db.purge_synced(days).await; - } - - async fn flush_pushes(&self, db: &TursoDb) { - let cfg = self.config.read().await.clone(); - let Some(webhook_url) = cfg.webhook_url() else { - return; - }; - let Some(token) = auth::get_access_token(&cfg) else { - return; - }; - let due = match db.list_due_pushes().await { - Ok(d) => d, - Err(e) => { - *self.last_error.write().await = Some(e); - return; - } - }; - for row in due { - match self.webhook.push(&cfg, &webhook_url, &token, &row).await { - Ok(()) => { - let _ = db.mark_synced(&row.id).await; - } - Err(e) => { - let _ = db - .mark_push_failed(&row.id, &e, row.retry_count + 1) - .await; - *self.last_error.write().await = Some(e); - } - } - } - } - pub async fn home_state(&self) -> HomeState { let cfg = self.config.read().await.clone(); - let mut turso_ok = false; - let mut pending_count = 0; - let mut active = None; - let mut history = Vec::new(); - match self.ensure_db().await { - Ok(db) => { - turso_ok = db.ping().await.is_ok(); - pending_count = db.count_pending().await.unwrap_or(0); - active = db.get_active().await.ok().flatten(); - history = db.list_sessions(100).await.unwrap_or_default(); - } - Err(e) => { - *self.last_error.write().await = Some(e); - } - } + let poll = cfg.poll_interval_secs.max(1); + let view = self.db_view.read().await.clone(); + let live = self.live.read().await.clone(); + let health = self.health.read().await.clone(); + let last_error = health + .last_error + .clone() + .or(self.last_error.read().await.clone()); + let active = overlay_active(&live, view.active.clone()); let sync = SyncStatus { - turso_ok, - pending_count, - last_error: self.last_error.read().await.clone(), + turso_ok: view.turso_ok, + pending_count: view.pending_count, + last_error, active_title: active.as_ref().map(|a| a.title.clone()), webhook_configured: cfg.webhook_url().is_some() && auth::get_access_token(&cfg).is_some(), + last_tick_at: health.last_detect_at.map(|t| t.to_rfc3339()), + loop_alive: health.loop_alive(poll), + db_reconnects: health.db_reconnects, + detect_timeouts: health.detect_timeouts, }; HomeState { sync, active, - history, + history: view.history, pending_detections: self.pending_detections.read().await.clone(), } } @@ -319,55 +222,92 @@ impl AppState { title: title.clone(), identity_id: identity_id.clone(), }; - if let Some(db) = self.db.read().await.as_ref() { - db.upsert_mapping(&mapping.fingerprint, &mapping.title, &mapping.identity_id) - .await?; - // Confirming implies tracking — clear any prior ignore. - let _ = db.remove_ignored(&identity_id).await; - } { let mut pipe = self.pipeline.write().await; pipe.ignored_identities.remove(&identity_id); - pipe.user_mappings.insert(fingerprint, mapping); + pipe.user_mappings.insert(fingerprint.clone(), mapping); } + self.persist_or_db( + |reply| PersistCmd::Confirm { + fingerprint: fingerprint.clone(), + title: title.clone(), + reply, + }, + || async { + if let Some(db) = self.db.read().await.as_ref() { + db.upsert_mapping(&fingerprint, &title, &identity_id) + .await?; + let _ = db.remove_ignored(&identity_id).await; + } + Ok(()) + }, + ) + .await?; Ok(()) } pub async fn ignore_game(&self, identity_id: String, title: String) -> Result<(), String> { - if let Some(db) = self.db.read().await.as_ref() { - db.upsert_ignored(&identity_id, &title).await?; - // Don't track ⇒ drop any in-flight session without pushing. - let actives = db.list_active().await.unwrap_or_default(); - let discard_ids: Vec = actives - .into_iter() - .filter(|s| s.identity_id == identity_id) - .map(|s| s.id) - .collect(); - if !discard_ids.is_empty() { - let _ = db.discard_active_sessions(&discard_ids).await; - } - let prev = self.active_identity_id.read().await.clone(); - if prev.as_deref() == Some(identity_id.as_str()) { - *self.active_identity_id.write().await = None; + { + let mut live = self.live.write().await; + if live.identity_id() == Some(identity_id.as_str()) { + live.clear_identity(); } } self.pipeline .write() .await .ignored_identities - .insert(identity_id); + .insert(identity_id.clone()); + self.ignored_titles + .write() + .await + .insert(identity_id.clone(), title.clone()); + self.persist_or_db( + |reply| PersistCmd::Ignore { + identity_id: identity_id.clone(), + title: title.clone(), + reply, + }, + || async { + if let Some(db) = self.db.read().await.as_ref() { + db.upsert_ignored(&identity_id, &title).await?; + let actives = db.list_active().await.unwrap_or_default(); + let discard_ids: Vec = actives + .into_iter() + .filter(|s| s.identity_id == identity_id) + .map(|s| s.id) + .collect(); + if !discard_ids.is_empty() { + let _ = db.discard_active_sessions(&discard_ids).await; + } + } + Ok(()) + }, + ) + .await?; Ok(()) } pub async fn unignore_game(&self, identity_id: String) -> Result<(), String> { - if let Some(db) = self.db.read().await.as_ref() { - db.remove_ignored(&identity_id).await?; - } self.pipeline .write() .await .ignored_identities .remove(&identity_id); + self.ignored_titles.write().await.remove(&identity_id); + self.persist_or_db( + |reply| PersistCmd::Unignore { + identity_id: identity_id.clone(), + reply, + }, + || async { + if let Some(db) = self.db.read().await.as_ref() { + db.remove_ignored(&identity_id).await?; + } + Ok(()) + }, + ) + .await?; Ok(()) } @@ -393,90 +333,67 @@ impl AppState { let identity_id = steam_app_id .map(|sid| format!("steam:{sid}")) .unwrap_or_else(|| format!("manual:{}", game.id)); - if let Some(db) = self.db.read().await.as_ref() { - db.upsert_manual_game(&game).await?; - let _ = db.remove_ignored(&identity_id).await; - } { let mut pipe = self.pipeline.write().await; pipe.ignored_identities.remove(&identity_id); - // Replace existing entry with same exe+hint if present. - pipe.manual_games - .retain(|g| !(g.exe_name.eq_ignore_ascii_case(&game.exe_name) - && g.path_hint == game.path_hint)); + pipe.manual_games.retain(|g| { + !(g.exe_name.eq_ignore_ascii_case(&game.exe_name) && g.path_hint == game.path_hint) + }); pipe.manual_games.push(game.clone()); } + self.persist_or_db( + |reply| PersistCmd::AddManual { + game: game.clone(), + identity_id: identity_id.clone(), + reply, + }, + || async { + if let Some(db) = self.db.read().await.as_ref() { + db.upsert_manual_game(&game).await?; + let _ = db.remove_ignored(&identity_id).await; + } + Ok(()) + }, + ) + .await?; Ok(game) } } -/// DB-authoritative session reconcile. Call every tick so process restarts cannot -/// leave orphan `active` rows or open duplicates. -pub async fn reconcile_active_sessions( - db: &TursoDb, - primary: Option<&GameIdentity>, - prev_identity_id: Option<&str>, -) -> Result<(), String> { - let actives = db.list_active().await?; - - match primary { - Some(identity) => { - // End (push) actives for other identities — real switch / leftover. - for row in actives.iter().filter(|s| s.identity_id != identity.id) { - db.end_session(&row.id).await?; - } - - let mut same: Vec<_> = actives - .into_iter() - .filter(|s| s.identity_id == identity.id) - .collect(); - // list_active is oldest-first, but sort defensively. - same.sort_by_key(|s| s.started_at); - - if same.is_empty() { - db.open_session(identity).await?; - } else { - // Keep oldest continuous session; discard restart duplicates. - let discard_ids: Vec = same.iter().skip(1).map(|s| s.id.clone()).collect(); - if !discard_ids.is_empty() { - db.discard_active_sessions(&discard_ids).await?; - } - } - } - None => { - if prev_identity_id.is_some() { - // Monitor was tracking something; game quit → end and push. - for row in &actives { - db.end_session(&row.id).await?; - } - } else { - // Cold start / restart with nothing running → discard orphans (no webhook spam). - let ids: Vec = actives.into_iter().map(|s| s.id).collect(); - if !ids.is_empty() { - tracing::warn!( - count = ids.len(), - "discarding orphaned active sessions after cold idle reconcile" - ); - db.discard_active_sessions(&ids).await?; - } - } - } +fn overlay_active(live: &LiveSession, db_active: Option) -> Option { + let Some(identity) = live.identity.as_ref() else { + return db_active; + }; + if db_active + .as_ref() + .is_some_and(|a| a.identity_id == identity.id) + { + return db_active; } - Ok(()) + Some(SessionRow { + id: live + .db_session_id + .clone() + .unwrap_or_else(|| format!("live:{}", identity.id)), + identity_id: identity.id.clone(), + title: identity.title.clone(), + steam_app_id: identity.steam_app_id, + exe: identity.exe.clone(), + source: identity.source.clone(), + started_at: live.started_at.unwrap_or_else(chrono::Utc::now), + ended_at: None, + duration_secs: None, + push_status: PushStatus::Active, + acked_at: None, + retry_count: 0, + next_retry_at: None, + last_error: None, + }) } + pub async fn list_trackable_games(state: &AppState) -> Vec { - let ignored_titles: std::collections::HashMap = - if let Some(db) = state.db.read().await.as_ref() { - db.list_ignored() - .await - .unwrap_or_default() - .into_iter() - .map(|i| (i.identity_id, i.title)) - .collect() - } else { - Default::default() - }; + let ignored_titles = state.ignored_titles.read().await.clone(); let pipe = state.pipeline.read().await; let ignored = &pipe.ignored_identities; @@ -559,8 +476,10 @@ pub async fn list_trackable_games(state: &AppState) -> Vec { #[cfg(test)] mod tests { use super::*; - use crate::db::{PushStatus, TursoDb}; + use crate::db::TursoDb; use crate::identity::{Confidence, GameIdentity}; + use crate::live_session::{DetectSample, LiveSession}; + use crate::persist::flush_live; use chrono::{Duration, Utc}; use tempfile::tempdir; @@ -576,105 +495,6 @@ mod tests { } } - #[tokio::test] - async fn reconcile_keeps_oldest_discards_duplicates_when_still_playing() { - let dir = tempdir().unwrap(); - let db = TursoDb::open(dir.path().join("r1.db")).await.unwrap(); - let apex = identity("steam:1172470", "Apex"); - let older = db - .force_insert_active(&apex, Utc::now() - Duration::minutes(20)) - .await - .unwrap(); - let newer = db - .force_insert_active(&apex, Utc::now() - Duration::minutes(5)) - .await - .unwrap(); - - reconcile_active_sessions(&db, Some(&apex), None) - .await - .unwrap(); - - let actives = db.list_active().await.unwrap(); - assert_eq!(actives.len(), 1); - assert_eq!(actives[0].id, older.id); - assert!(db.get_session(&newer.id).await.unwrap().is_none()); - assert!(db.list_due_pushes().await.unwrap().is_empty()); - } - - #[tokio::test] - async fn reconcile_ends_other_identity_and_opens_current() { - let dir = tempdir().unwrap(); - let db = TursoDb::open(dir.path().join("r2.db")).await.unwrap(); - let a = identity("steam:1", "A"); - let b = identity("steam:2", "B"); - let old = db.open_session(&a).await.unwrap(); - - reconcile_active_sessions(&db, Some(&b), Some("steam:1")) - .await - .unwrap(); - - let ended = db.get_session(&old.id).await.unwrap().unwrap(); - assert_eq!(ended.push_status, PushStatus::Pending); - assert!(ended.ended_at.is_some()); - - let actives = db.list_active().await.unwrap(); - assert_eq!(actives.len(), 1); - assert_eq!(actives[0].identity_id, "steam:2"); - } - - #[tokio::test] - async fn reconcile_cold_idle_discards_orphans_without_push() { - let dir = tempdir().unwrap(); - let db = TursoDb::open(dir.path().join("r3.db")).await.unwrap(); - let apex = identity("steam:1172470", "Apex"); - db.force_insert_active(&apex, Utc::now() - Duration::hours(1)) - .await - .unwrap(); - db.force_insert_active(&apex, Utc::now() - Duration::minutes(30)) - .await - .unwrap(); - - reconcile_active_sessions(&db, None, None).await.unwrap(); - - assert!(db.list_active().await.unwrap().is_empty()); - assert!(db.list_due_pushes().await.unwrap().is_empty()); - assert!(db.list_sessions(10).await.unwrap().is_empty()); - } - - /// Game quit (`primary = None`) ends and queues push on that tick — no end-grace. - #[tokio::test] - async fn reconcile_warm_idle_ends_and_queues_push() { - let dir = tempdir().unwrap(); - let db = TursoDb::open(dir.path().join("r4.db")).await.unwrap(); - let apex = identity("steam:1172470", "Apex"); - let row = db.open_session(&apex).await.unwrap(); - - reconcile_active_sessions(&db, None, Some("steam:1172470")) - .await - .unwrap(); - - assert!(db.list_active().await.unwrap().is_empty()); - let due = db.list_due_pushes().await.unwrap(); - assert_eq!(due.len(), 1); - assert_eq!(due[0].id, row.id); - assert_eq!(due[0].push_status, PushStatus::Pending); - } - - #[tokio::test] - async fn reconcile_opens_when_playing_with_no_active() { - let dir = tempdir().unwrap(); - let db = TursoDb::open(dir.path().join("r5.db")).await.unwrap(); - let apex = identity("steam:1172470", "Apex"); - - reconcile_active_sessions(&db, Some(&apex), None) - .await - .unwrap(); - - let actives = db.list_active().await.unwrap(); - assert_eq!(actives.len(), 1); - assert_eq!(actives[0].identity_id, "steam:1172470"); - } - #[tokio::test] async fn ignore_discards_all_actives_without_push() { let state = AppState::new(); @@ -690,7 +510,7 @@ mod tests { .await .unwrap(); *state.db.write().await = Some(std::sync::Arc::new(db)); - *state.active_identity_id.write().await = Some("steam:1172470".into()); + state.live.write().await.identity = Some(apex); state .ignore_game("steam:1172470".into(), "Apex".into()) @@ -702,7 +522,7 @@ mod tests { assert!(db.get_session(&older.id).await.unwrap().is_none()); assert!(db.get_session(&newer.id).await.unwrap().is_none()); assert!(db.list_due_pushes().await.unwrap().is_empty()); - assert!(state.active_identity_id.read().await.is_none()); + assert!(state.live.read().await.identity.is_none()); assert!(state .pipeline .read() @@ -712,19 +532,48 @@ mod tests { } #[tokio::test] - async fn ensure_db_opens_missing_handle() { + async fn detect_updates_live_when_persist_is_offline() { let state = AppState::new(); + let t0 = Utc::now(); + let sample = DetectSample { + observed_at: t0, + primary: Some(identity("steam:1", "RL")), + }; + state.live.write().await.apply(&sample); + let home = state.home_state().await; + assert_eq!(home.active.as_ref().map(|a| a.identity_id.as_str()), Some("steam:1")); + assert!(!home.sync.turso_ok); + state.live.write().await.apply(&DetectSample { + observed_at: t0 + Duration::seconds(9), + primary: None, + }); + assert_eq!(state.live.read().await.pending_ends.len(), 1); + } + + #[tokio::test] + async fn overlay_shows_live_when_db_view_empty() { + let mut live = LiveSession::default(); + live.apply(&DetectSample { + observed_at: Utc::now(), + primary: Some(identity("steam:1", "RL")), + }); + let row = overlay_active(&live, None).unwrap(); + assert_eq!(row.identity_id, "steam:1"); + assert_eq!(row.push_status, crate::db::PushStatus::Active); + } + + #[tokio::test] + async fn flush_live_opens_when_playing() { let dir = tempdir().unwrap(); - let path = dir.path().join("ensure.db"); - { - let mut cfg = state.config.write().await; - cfg.db_path = Some(path.to_string_lossy().to_string()); - } - assert!(state.db.read().await.is_none()); - let db = state.ensure_db().await.expect("ensure"); - assert!(db.ping().await.is_ok()); - // Second call reuses the live handle. - let db2 = state.ensure_db().await.expect("ensure again"); - assert!(db2.ping().await.is_ok()); + let db = TursoDb::open(dir.path().join("open.db")).await.unwrap(); + let mut live = LiveSession::default(); + live.apply(&DetectSample { + observed_at: Utc::now(), + primary: Some(identity("steam:1172470", "Apex")), + }); + flush_live(&db, &mut live).await.unwrap(); + let actives = db.list_active().await.unwrap(); + assert_eq!(actives.len(), 1); + assert_eq!(actives[0].identity_id, "steam:1172470"); } } diff --git a/src/App.css b/src/App.css index 0bf2035..b944090 100644 --- a/src/App.css +++ b/src/App.css @@ -858,11 +858,36 @@ code { } .settings-advanced[open] summary { - margin-bottom: 0.35rem; + margin-bottom: 0.15rem; } -.settings-advanced .field:first-of-type { - margin-top: 0.5rem; +.advanced-group { + margin-top: 0.7rem; + padding-top: 0.7rem; + border-top: 1px solid var(--line); +} + +.advanced-group:first-of-type { + margin-top: 0.25rem; + padding-top: 0; + border-top: none; +} + +.advanced-group-label { + margin: 0 0 0.45rem; + font-family: var(--font-mono); + font-size: 0.6875rem; + text-transform: uppercase; + letter-spacing: 0.14em; + color: var(--faint); +} + +.advanced-group .field { + margin-top: 0.65rem; +} + +.advanced-group .field:first-of-type { + margin-top: 0; } .settings-save { @@ -994,6 +1019,21 @@ code { margin-top: 0.85rem; } +.path-row { + display: flex; + gap: 0.45rem; + align-items: stretch; +} + +.path-row input { + flex: 1; + min-width: 0; +} + +.path-row .btn { + flex-shrink: 0; +} + .dialog-actions { margin-top: 1.15rem; justify-content: flex-end; diff --git a/src/App.tsx b/src/App.tsx index 0cbc416..c80447b 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -1,6 +1,7 @@ import { invoke } from "@tauri-apps/api/core"; import { listen } from "@tauri-apps/api/event"; -import { useCallback, useEffect, useState } from "react"; +import { open } from "@tauri-apps/plugin-dialog"; +import { useCallback, useEffect, useRef, useState } from "react"; import { QMark } from "./components/QMark"; import { Settings, @@ -34,6 +35,10 @@ interface SyncStatus { lastError?: string; activeTitle?: string; webhookConfigured: boolean; + lastTickAt?: string; + loopAlive?: boolean; + dbReconnects?: number; + detectTimeouts?: number; } interface PendingDetection { @@ -59,6 +64,20 @@ interface TrackableGame { trackingEnabled: boolean; } +async function invokeTimeout(cmd: string, ms = 4000): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + invoke(cmd), + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`${cmd} timed out`)), ms); + }), + ]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} + function BrandMark({ size = "sm" }: { size?: "sm" | "md" }) { const px = size === "md" ? 36 : 28; return ( @@ -143,23 +162,32 @@ function App() { setMessage(text); }, []); + const lastTimeoutToast = useRef(null); + const elapsed = useElapsedSecs(home?.active?.startedAt); /** Live status only — never overwrite draft settings while the user is typing. */ const refresh = useCallback(async () => { try { const [h, a, o, g] = await Promise.all([ - invoke("get_home"), - invoke("get_auth_state"), - invoke("is_onboarded"), - invoke("list_games"), + invokeTimeout("get_home"), + invokeTimeout("get_auth_state"), + invokeTimeout("is_onboarded"), + invokeTimeout("list_games"), ]); setHome(h); setAuth(a); setOnboarded(o); setGames(g); + lastTimeoutToast.current = null; } catch (e) { - showToast(String(e), true); + const text = String(e); + const isTimeout = text.includes("timed out"); + if (isTimeout && lastTimeoutToast.current === text) { + return; + } + lastTimeoutToast.current = isTimeout ? text : null; + showToast(text, true); } }, [showToast]); @@ -175,9 +203,11 @@ function App() { void loadConfig(); void refresh(); const unsubs: Array<() => void> = []; - listen("qmonitor://tick", () => { - refresh(); - }).then((fn) => unsubs.push(fn)); + const onTick = () => { + if (typeof document !== "undefined" && document.hidden) return; + void refresh(); + }; + listen("qmonitor://tick", onTick).then((fn) => unsubs.push(fn)); listen("qmonitor://auth-success", async () => { setLoginPhase("idle"); setShowManualAuth(false); @@ -188,9 +218,16 @@ function App() { listen("qmonitor://auth-waiting", () => { setLoginPhase("waiting"); }).then((fn) => unsubs.push(fn)); - const id = setInterval(refresh, 5000); + const onVis = () => { + if (!document.hidden) void refresh(); + }; + document.addEventListener("visibilitychange", onVis); + const id = setInterval(() => { + if (!document.hidden) void refresh(); + }, 5000); return () => { clearInterval(id); + document.removeEventListener("visibilitychange", onVis); unsubs.forEach((u) => u()); }; }, [refresh, loadConfig, showToast]); @@ -292,6 +329,30 @@ function App() { setAddOpen(true); } + function titleFromExePath(path: string): string { + const base = path.replace(/[/\\]+$/, "").split(/[/\\]/).pop() ?? ""; + return base.replace(/\.exe$/i, "").trim(); + } + + async function browseExe() { + try { + const selected = await open({ + multiple: false, + directory: false, + title: "Choose game executable", + filters: [ + { name: "Executables", extensions: ["exe"] }, + { name: "All files", extensions: ["*"] }, + ], + }); + if (typeof selected !== "string" || !selected) return; + setAddExe(selected); + setAddTitle((current) => current.trim() || titleFromExePath(selected)); + } catch (e) { + showToast(String(e), true); + } + } + async function submitAddGame() { const title = addTitle.trim(); const exePath = addExe.trim(); @@ -679,7 +740,14 @@ function App() {