diff --git a/docs/architecture/canonical-session.md b/docs/architecture/canonical-session.md index e2ab3892..33b1d91d 100644 --- a/docs/architecture/canonical-session.md +++ b/docs/architecture/canonical-session.md @@ -106,7 +106,7 @@ Workflow-tool spawns). Each line is a JSON object discriminated by |---|---| | `ID` | main sessions: `` basename of the JSONL file (UUID), matching the `sessionId` field inside records. Subagents: the filename stem (`agent-`) — their records carry the parent's `sessionId`, so the stem is the only stable identity | | `Agent` | constant `"claude-code"` | -| `DeviceID` | constant `"local"` in cut 1 (fingerprint lands with sync) | +| `DeviceID` | `device.IDOnce()` | | `ProjectPath` | first non-empty `cwd` encountered on any record | | `StartedAt` | `min(timestamp)` over all records (RFC 3339, UTC after parse) | | `LastActivityAt` | `max(timestamp)` over all records | @@ -161,7 +161,7 @@ The full envelope reference is `docs/sources/codex.md`. |---|---| | `ID` | envelope: `session_meta.payload.id`. Legacy/missing meta: UUID suffix of the filename (`...-.jsonl`) | | `Agent` | constant `"codex"` | -| `DeviceID` | constant `"local"` in cut 2 | +| `DeviceID` | `device.IDOnce()` | | `ProjectPath` | first non-empty of `session_meta.payload.cwd`, then `turn_context.payload.cwd` | | `StartedAt` / `LastActivityAt` | `min`/`max` of every record's top-level `timestamp` (both envelope and legacy carry it) | | `FirstPrompt` | first `response_item.payload.type=="message"` with `role=="user"` and non-empty `content[*].input_text` (legacy: first `{type:"message", role:"user"}`); whitespace-collapsed + truncated to 200 runes | diff --git a/docs/architecture/cli.md b/docs/architecture/cli.md index ae8d826e..5351f953 100644 --- a/docs/architecture/cli.md +++ b/docs/architecture/cli.md @@ -33,7 +33,7 @@ is the Connect-Go client in `internal/cli` that talks to `prosa-server`. | `prosa devices …` | `internal/cli/devices.go` | `list`, `rename`, `revoke` | | `prosa profiles …` | `internal/cli/profiles.go` | `list`, `add`, `remove`, `set-path` — manage per-agent import locations (local `profiles.json`) | | `prosa schedule …` | `internal/cli/schedule_cmd.go` + `internal/cli/schedule/` | `install`, `status`, `uninstall` | -| `prosa prune` | `internal/cli/prune.go` | Delete local raw copies of old, server-confirmed sessions (`--older-than`, `--dry-run`, `--limit`) | +| `prosa prune` | `internal/cli/prune.go` | Delete local raw copies of old sessions a prior sync confirmed (`pushed_hash` equals `raw_hash`). `--older-than`, `--dry-run`, `--limit`. When none qualify, reports how many older sessions still lack that confirmation. | | `prosa setup` | `internal/cli/setup.go` | Wizard wrapping login + schedule + first sync | Each handler: @@ -132,9 +132,11 @@ Each query funnels into a small set of functions: | `WriteSession`, `LastHash` | `sync` (importer `Sink`) | | `ListDevicesMap` | timeline column resolution | | `Analytics*` | `analytics` | -| `RebindLocalSessions` | one-time migration helper during `setup` | +| `RebindLocalSessions` | `sync` and `prune`: move seed `local` sessions onto this machine's fingerprint | +| `RebindDevicesByMachineID` | `sync` and `prune`: collapse other device rows that share this machine id | | `RecordPushed`, `RecordPushedBatch` | `sync` (push confirmation + manifest backfill) | | `ListPruneCandidates`, `MarkPruned`, `ClearPruned` | `prune` | +| `CountOldUnconfirmed` | `prune`, when no candidate is confirmed | | `PruneAdvisory` | `sync` (post-run Prune summary line) | The CLI never builds SQL by hand. If a query doesn't exist, add a function diff --git a/docs/architecture/store.md b/docs/architecture/store.md index 44118b37..3912ad5c 100644 --- a/docs/architecture/store.md +++ b/docs/architecture/store.md @@ -111,14 +111,17 @@ down is for manual recovery. | Column | Type | Notes | | --- | --- | --- | -| `id` | TEXT PRIMARY KEY | Device fingerprint | -| `hostname` | TEXT | | -| `machine_id` | TEXT | OS-level machine ID | +| `id` | TEXT PRIMARY KEY | `hex(sha256(hostname + NUL + machine-id))[:16]` | +| `hostname` | TEXT | LocalHostName on macOS; system hostname elsewhere, with a trailing `.local` removed | +| `machine_id` | TEXT | IOPlatformUUID on macOS; `/etc/machine-id` on Linux | | `friendly_name` | TEXT | Defaults to hostname; user-editable cross-device | | `fingerprinted_at` | TEXT | RFC3339 of first registration | -Seeded with one `'local'` row at install. Replaced by the real -fingerprint on first successful `prosa sync` via `RebindLocalSessions`. +Seeded with one `local` row at install. `prosa sync` and `prosa prune` +upsert the current fingerprint. `RebindLocalSessions` moves +`device_id = 'local'` sessions onto it. `RebindDevicesByMachineID` moves +sessions off every other row with the same non-empty `machine_id` and +deletes those rows. Rows with an empty `machine_id` stay in place. ### `sessions` @@ -251,10 +254,12 @@ Selected functions (full list in `internal/store/`): | `RecordPushed(ctx, sessionID, hash, uri)` | Persist a server-confirmed push into `sync_state` | | `RecordPushedBatch(ctx, stamps)` | Manifest-driven backfill of pushed state, keeping each entry's `last_synced_at` | | `ListPruneCandidates(ctx, deviceID, before, limit)` | Pushed, unpruned sessions inactive since `before` | +| `CountOldUnconfirmed(ctx, deviceID, before)` | Older unpruned sessions whose `pushed_hash` is missing or differs from `raw_hash` | | `MarkPruned(ctx, sessionID, rawHash)` / `ClearPruned(ctx, sessionID)` | Guarded prune flip and its revert | | `PruneAdvisory(ctx, deviceID, before, pushedBefore)` | Count + bytes behind the sync summary's Prune line | | `ListDevicesMap(ctx)` | `id → friendly_name` lookup | -| `RebindLocalSessions(ctx, deviceID)` | Migrate `local` seed device | +| `RebindLocalSessions(ctx, deviceID)` | Move `device_id = 'local'` sessions onto a fingerprint | +| `RebindDevicesByMachineID(ctx, device)` | Move sessions off other rows with the same machine id, then delete those rows | | `Analytics*` | Per-report queries | The `SessionFilter` type carries the parsed `--since/--until/--project/...` diff --git a/docs/concepts.md b/docs/concepts.md index aef384be..f3696ff1 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -80,6 +80,12 @@ its metadata, turns, FTS entries, and analytics locally; only raw reads re-import of a changed source file rewrites the raw copy and un-prunes the session. +`prosa prune` lists only sessions a prior sync confirmed: `pushed_hash` +equals the current `raw_hash`, the row is not pruned, and last activity is +older than the window. Older sessions that still lack that confirmation stay +on disk. Prune reports their count and names `prosa sync`, which records the +confirmation from the server manifest. + ## Project identity prosa picks a project for each session in this order: @@ -101,9 +107,17 @@ meaningful. A *device* is a single machine prosa has been set up on. Each device has a fingerprint and a `friendly_name`. -Fingerprint = `hash(hostname + machine-id)`. Sources of the machine-id: +The fingerprint is `hex(sha256(hostname + NUL + machine-id))[:16]`. + +On macOS the hostname is the LocalHostName (`scutil --get LocalHostName`, +trimmed). When that command fails or returns empty, prosa uses the system +hostname with a trailing `.local` removed. On Linux, and on every other +platform, the hostname is the system hostname (`os.Hostname()`) with a +trailing `.local` removed. -- **Linux**: `/etc/machine-id`. +Machine-id sources: + +- **Linux**: `/etc/machine-id`, then `/var/lib/dbus/machine-id`. - **macOS**: `IOPlatformUUID`. - **Windows**: `MachineGuid` from the registry. (Not in MVP target platforms.) @@ -112,6 +126,12 @@ The `friendly_name` defaults to the hostname. You can rename via `prosa devices rename ` (cross-device only, talks to the server). +`prosa sync` and `prosa prune` upsert this machine's device row. They move +sessions still stored on the seed `local` id onto the current fingerprint, +and they collapse other local device rows that share its machine-id: +sessions are reassigned to the current fingerprint and the extra rows are +deleted. Rows with an empty machine-id stay in place. + The CLI sends its device fingerprint as part of every authenticated request. The server uses this to attribute pushed sessions to a device. diff --git a/internal/cli/device_bind.go b/internal/cli/device_bind.go new file mode 100644 index 00000000..e45b9208 --- /dev/null +++ b/internal/cli/device_bind.go @@ -0,0 +1,41 @@ +package cli + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/c3-oss/prosa/internal/device" + "github.com/c3-oss/prosa/internal/store" +) + +// bindLocalDevice upserts this machine's fingerprint, moves sessions still +// tagged with the seed id onto it, and collapses other device rows that +// share its machine id. sync and prune both call it before they read by +// device id. +func bindLocalDevice(ctx context.Context, s *store.Store) (store.Device, error) { + dev := store.Device{ + ID: device.IDOnce(), + Hostname: device.Hostname(), + MachineID: device.MachineID(), + FriendlyName: device.FriendlyName(), + FingerprintedAt: time.Now().UTC(), + } + if err := s.UpsertDevice(ctx, dev); err != nil { + return store.Device{}, fmt.Errorf("upsert device: %w", err) + } + if n, err := s.RebindLocalSessions(ctx, dev.ID); err != nil { + return store.Device{}, fmt.Errorf("rebind 'local' sessions to %s: %w", dev.ID, err) + } else if n > 0 { + slog.Info("rebound legacy 'local' sessions to fingerprint", + "device_id", dev.ID, "rows", n) + } + if n, err := s.RebindDevicesByMachineID(ctx, dev); err != nil { + return store.Device{}, fmt.Errorf("rebind devices by machine id: %w", err) + } else if n > 0 { + slog.Info("rebound sessions from stale device ids", + "device_id", dev.ID, "rows", n) + } + return dev, nil +} diff --git a/internal/cli/prune.go b/internal/cli/prune.go index 7d38128a..1d1297db 100644 --- a/internal/cli/prune.go +++ b/internal/cli/prune.go @@ -15,7 +15,6 @@ import ( "github.com/c3-oss/prosa/internal/cli/render" "github.com/c3-oss/prosa/internal/cli/rpc" - "github.com/c3-oss/prosa/internal/device" "github.com/c3-oss/prosa/internal/importers/importerutil" "github.com/c3-oss/prosa/internal/paths" "github.com/c3-oss/prosa/internal/rawlock" @@ -75,6 +74,7 @@ type pruneJSONSummary struct { Errors int `json:"errors"` ReclaimedBytes int64 `json:"reclaimed_bytes"` DryRun bool `json:"dry_run"` + Unconfirmed int `json:"unconfirmed,omitempty"` } func runPrune(cmd *cobra.Command, _ []string) error { @@ -103,18 +103,33 @@ func runPrune(cmd *cobra.Command, _ []string) error { } defer func() { _ = s.Close() }() + dev, err := bindLocalDevice(ctx, s) + if err != nil { + return err + } + before := time.Now().UTC().Add(-olderThan) // Limit is applied after the server confirms a row. A SQL limit would // let unconfirmed old sessions consume the whole cap. - candidates, err := s.ListPruneCandidates(ctx, device.IDOnce(), before, 0) + candidates, err := s.ListPruneCandidates(ctx, dev.ID, before, 0) if err != nil { return fmt.Errorf("list prune candidates: %w", err) } enc := json.NewEncoder(os.Stdout) if len(candidates) == 0 { + unconfirmed, err := s.CountOldUnconfirmed(ctx, dev.ID, before) + if err != nil { + return err + } if g.JSON { - return enc.Encode(pruneJSONSummary{Type: "summary", DryRun: pruneDryRunFlag}) + return enc.Encode(pruneJSONSummary{ + Type: "summary", DryRun: pruneDryRunFlag, Unconfirmed: unconfirmed, + }) + } + if unconfirmed > 0 { + fmt.Fprintf(os.Stderr, "Nothing to prune. Unconfirmed sessions older than the window: %d; run `prosa sync` first.\n", unconfirmed) + return nil } fmt.Fprintln(os.Stderr, "Nothing to prune.") return nil diff --git a/internal/cli/prune_test.go b/internal/cli/prune_test.go index 45673986..bdac3457 100644 --- a/internal/cli/prune_test.go +++ b/internal/cli/prune_test.go @@ -342,6 +342,91 @@ func TestPruneDryRunConfirmsWithServer(t *testing.T) { require.True(t, sawSkip) } +func TestPruneDryRunFindsSessionsReboundFromStaleDevice(t *testing.T) { + const ( + pinnedID = "aabbccddeeff0011" + pinnedMid = "machine-pinned" + staleID = "stale-dhcp" + ) + restore := device.SetResolveForTest(pinnedID, "tbox", "tbox", pinnedMid) + t.Cleanup(restore) + + s := openPruneCommandStore(t) + ctx := context.Background() + now := time.Now().UTC() + require.Equal(t, pinnedID, device.IDOnce()) + require.NoError(t, s.UpsertDevice(ctx, store.Device{ + ID: staleID, + Hostname: "192.168.0.19", + MachineID: pinnedMid, + FriendlyName: "192.168.0.19", + FingerprintedAt: now.Add(-time.Hour), + })) + rawPath, hash := seedCommandSession(t, s, "stale-sess", 60*24*time.Hour, []byte("raw-stale")) + _, err := s.DB().ExecContext(ctx, `UPDATE sessions SET device_id = ? WHERE id = ?`, staleID, "stale-sess") + require.NoError(t, err) + + before := time.Now().UTC().Add(-30 * 24 * time.Hour) + got, err := s.ListPruneCandidates(ctx, device.IDOnce(), before, 0) + require.NoError(t, err) + require.Empty(t, got) + + restorePruneHooks(t) + loadPruneManifest = func(context.Context, string, string) (map[string]serverManifestRow, error) { + return confirmedManifest("stale-sess", hash), nil + } + + stdout, _ := captureStdoutStderr(t, func() { + cmd := newRootCmd() + cmd.SetArgs([]string{"--json", "prune", "--dry-run"}) + require.NoError(t, cmd.Execute()) + }) + require.Contains(t, stdout, `"session_id":"stale-sess"`) + require.Contains(t, stdout, `"status":"would_prune"`) + require.FileExists(t, rawPath) + + var deviceID string + require.NoError(t, s.DB().QueryRowContext(ctx, + `SELECT device_id FROM sessions WHERE id = 'stale-sess'`).Scan(&deviceID)) + require.Equal(t, pinnedID, deviceID) + var staleRows int + require.NoError(t, s.DB().QueryRowContext(ctx, + `SELECT COUNT(*) FROM devices WHERE id = ?`, staleID).Scan(&staleRows)) + require.Zero(t, staleRows) +} + +func TestPruneNothingHintsUnconfirmed(t *testing.T) { + s := openPruneCommandStore(t) + ctx := context.Background() + rawPath, _ := seedCommandSession(t, s, "old-null", 60*24*time.Hour, []byte("raw-null")) + _, err := s.DB().ExecContext(ctx, `UPDATE sync_state SET pushed_hash = NULL WHERE session_id = ?`, "old-null") + require.NoError(t, err) + seedCommandSession(t, s, "old-diverged", 50*24*time.Hour, []byte("raw-diverged")) + _, err = s.DB().ExecContext(ctx, `UPDATE sync_state SET pushed_hash = ? WHERE session_id = ?`, "other-hash", "old-diverged") + require.NoError(t, err) + seedCommandSession(t, s, "recent-null", time.Hour, []byte("raw-recent")) + _, err = s.DB().ExecContext(ctx, `UPDATE sync_state SET pushed_hash = NULL WHERE session_id = ?`, "recent-null") + require.NoError(t, err) + + restorePruneHooks(t) + _, stderr := captureStdoutStderr(t, func() { + cmd := newRootCmd() + cmd.SetArgs([]string{"prune", "--dry-run"}) + require.NoError(t, cmd.Execute()) + }) + require.Contains(t, stderr, "Nothing to prune. Unconfirmed sessions older than the window: 2; run `prosa sync` first.") + require.FileExists(t, rawPath) + + stdout, _ := captureStdoutStderr(t, func() { + cmd := newRootCmd() + cmd.SetArgs([]string{"--json", "prune", "--dry-run"}) + require.NoError(t, cmd.Execute()) + }) + require.Contains(t, stdout, `"unconfirmed":2`) + require.Contains(t, stdout, `"dry_run":true`) + require.NotContains(t, stdout, `"status":"would_prune"`) +} + func TestPruneDryRunErrorsWhenServerUnreachable(t *testing.T) { s := openPruneCommandStore(t) rawPath, _ := seedCommandSession(t, s, "s1", 60*24*time.Hour, []byte("raw")) diff --git a/internal/cli/sync.go b/internal/cli/sync.go index 0f704b26..a195be26 100644 --- a/internal/cli/sync.go +++ b/internal/cli/sync.go @@ -9,7 +9,6 @@ import ( "github.com/spf13/cobra" - "github.com/c3-oss/prosa/internal/device" "github.com/c3-oss/prosa/internal/importers/claudecode" "github.com/c3-oss/prosa/internal/importers/codex" "github.com/c3-oss/prosa/internal/importers/cursor" @@ -88,23 +87,9 @@ func runSync(cmd *cobra.Command, _ []string) error { } defer func() { _ = s.Close() }() - // Migrate any rows inserted under the seed `'local'` device id (v1 bundle - // restore path) to the real fingerprint — those sessions came from this machine. - dev := store.Device{ - ID: device.IDOnce(), - Hostname: device.Hostname(), - MachineID: device.MachineID(), - FriendlyName: device.FriendlyName(), - FingerprintedAt: time.Now().UTC(), - } - if err := s.UpsertDevice(ctx, dev); err != nil { - return fmt.Errorf("upsert device: %w", err) - } - if n, err := s.RebindLocalSessions(ctx, dev.ID); err != nil { - return fmt.Errorf("rebind 'local' sessions to %s: %w", dev.ID, err) - } else if n > 0 { - slog.Info("rebound legacy 'local' sessions to fingerprint", - "device_id", dev.ID, "rows", n) + dev, err := bindLocalDevice(ctx, s) + if err != nil { + return err } if err := backfillProjectIdentity(ctx, s); err != nil { diff --git a/internal/device/fingerprint.go b/internal/device/fingerprint.go index 2cf69885..76d5e065 100644 --- a/internal/device/fingerprint.go +++ b/internal/device/fingerprint.go @@ -1,12 +1,18 @@ // Package device resolves the stable per-machine identity prosa uses for -// the sessions.device_id column and the future server-side device row. -// The id is hex(sha256(hostname + machineID))[:16] so it survives -// hostname renames as long as the machine-id stays put, and survives -// machine-id changes (rare) as long as the hostname stays put. +// sessions.device_id and the server device row. // -// macOS: machineID = IOPlatformUUID parsed from `ioreg -rd1 -c IOPlatformExpertDevice`. -// Linux: machineID = /etc/machine-id (or /var/lib/dbus/machine-id fallback). -// Other: empty machineID; hostname alone still produces a stable id. +// The id is hex(sha256(hostname + "\x00" + machineID))[:16]. +// +// On darwin the hostname is `scutil --get LocalHostName`, trimmed. When +// scutil fails or returns empty, it is os.Hostname() with a trailing +// ".local" removed. On linux and every other platform it is os.Hostname() +// with a trailing ".local" removed. +// +// macOS machineID is the IOPlatformUUID parsed from +// `ioreg -rd1 -c IOPlatformExpertDevice`. +// Linux machineID is /etc/machine-id, or /var/lib/dbus/machine-id when +// that file is missing. +// Other platforms leave machineID empty; the hostname still yields an id. // // All public funcs return cached values after the first call so // repeated importer invocations on the same process pay the cost once. @@ -26,6 +32,13 @@ var ( once sync.Once cached resolved resolve = doResolve // overridable for tests + + // darwinLocalHostName is the raw stdout of `scutil --get LocalHostName`. + // Tests replace it so hostname resolution does not spawn a process. + darwinLocalHostName = execDarwinLocalHostName + + // osHostname is os.Hostname. Tests replace it. + osHostname = os.Hostname ) type resolved struct { @@ -45,8 +58,10 @@ func IDOnce() string { return cached.id } -// Hostname returns the machine hostname, with macOS' ".local" -// suffix stripped for friendliness. Cached. +// Hostname returns the hostname mixed into the fingerprint. Cached. +// On darwin this is the LocalHostName when scutil succeeds. Otherwise, +// and on every other platform, it is os.Hostname() with a trailing +// ".local" removed. func Hostname() string { once.Do(func() { cached = resolve() }) return cached.hostname @@ -66,16 +81,39 @@ func MachineID() string { return cached.machineID } -// Fingerprint computes hex(sha256(hostname + machineID))[:16]. Exposed so -// tests can reproduce the ID without going through the resolver. +// Fingerprint computes hex(sha256(hostname + "\x00" + machineID))[:16]. +// Exposed so tests can reproduce the id without going through the resolver. func Fingerprint(hostname, machineID string) string { h := sha256.Sum256([]byte(hostname + "\x00" + machineID)) return hex.EncodeToString(h[:])[:16] } +// SetResolveForTest pins the cached device identity. The returned function +// restores the previous resolver and clears the cache again. +func SetResolveForTest(id, hostname, friendly, machineID string) func() { + prev := resolve + resolve = func() resolved { + return resolved{ + id: id, + hostname: hostname, + friendly: friendly, + machineID: machineID, + } + } + resetResolveCache() + return func() { + resolve = prev + resetResolveCache() + } +} + +func resetResolveCache() { + once = sync.Once{} + cached = resolved{} +} + func doResolve() resolved { - host, _ := os.Hostname() - host = strings.TrimSuffix(host, ".local") + host := resolveHostname(runtime.GOOS) mid, err := readMachineID() return resolved{ id: Fingerprint(host, mid), @@ -86,6 +124,28 @@ func doResolve() resolved { } } +// resolveHostname picks the hostname mixed into the fingerprint. +// goos is injectable so tests cover darwin without running on darwin. +func resolveHostname(goos string) string { + if goos == "darwin" { + if raw, err := darwinLocalHostName(); err == nil { + if name := strings.TrimSpace(raw); name != "" { + return name + } + } + } + host, _ := osHostname() + return strings.TrimSuffix(host, ".local") +} + +func execDarwinLocalHostName() (string, error) { + out, err := exec.Command("scutil", "--get", "LocalHostName").Output() + if err != nil { + return "", err + } + return string(out), nil +} + func readMachineID() (string, error) { switch runtime.GOOS { case "darwin": diff --git a/internal/device/fingerprint_test.go b/internal/device/fingerprint_test.go index 6c6c2cb9..5a50c12a 100644 --- a/internal/device/fingerprint_test.go +++ b/internal/device/fingerprint_test.go @@ -1,11 +1,14 @@ package device import ( + "errors" "testing" "github.com/stretchr/testify/require" ) +var errScutil = errors.New("scutil failed") + func TestFingerprintIsStableAndShort(t *testing.T) { id1 := Fingerprint("laptop", "BEEFCAFE-1234") id2 := Fingerprint("laptop", "BEEFCAFE-1234") @@ -35,8 +38,82 @@ func TestFingerprintEmptyMachineIDStillStable(t *testing.T) { require.NotEmpty(t, a) } +func TestSetResolveForTestPinsMachineID(t *testing.T) { + restore := SetResolveForTest("aabbccddeeff0011", "tbox", "tbox", "mid-pin") + t.Cleanup(restore) + + require.Equal(t, "mid-pin", MachineID()) + require.Equal(t, "aabbccddeeff0011", IDOnce()) + require.Equal(t, "tbox", Hostname()) + require.Equal(t, "tbox", FriendlyName()) +} + func TestIDOnceMatchesFingerprintOfResolvedInputs(t *testing.T) { id := IDOnce() expected := Fingerprint(Hostname(), MachineID()) require.Equal(t, expected, id) } + +func TestResolveHostname(t *testing.T) { + origLocal := darwinLocalHostName + origHost := osHostname + t.Cleanup(func() { + darwinLocalHostName = origLocal + osHostname = origHost + }) + + tests := []struct { + name string + goos string + local func() (string, error) + host func() (string, error) + want string + scutil bool + }{ + { + name: "darwin scutil success", + goos: "darwin", + local: func() (string, error) { return " tbox\n", nil }, + host: func() (string, error) { return "192.168.0.19", nil }, + want: "tbox", + scutil: true, + }, + { + name: "darwin scutil failure falls back", + goos: "darwin", + local: func() (string, error) { return "", errScutil }, + host: func() (string, error) { return "192.168.0.19", nil }, + want: "192.168.0.19", + scutil: true, + }, + { + name: "darwin empty scutil falls back and strips .local", + goos: "darwin", + local: func() (string, error) { return " \n", nil }, + host: func() (string, error) { return "studio.local", nil }, + want: "studio", + scutil: true, + }, + { + name: "linux keeps os hostname and strips one .local", + goos: "linux", + local: func() (string, error) { return "ignored", nil }, + host: func() (string, error) { return "box.local", nil }, + want: "box", + scutil: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + called := false + darwinLocalHostName = func() (string, error) { + called = true + return tt.local() + } + osHostname = tt.host + require.Equal(t, tt.want, resolveHostname(tt.goos)) + require.Equal(t, tt.scutil, called) + }) + } +} diff --git a/internal/store/devices.go b/internal/store/devices.go index 881ce9f3..e9eaa5a7 100644 --- a/internal/store/devices.go +++ b/internal/store/devices.go @@ -97,6 +97,46 @@ func (s *Store) ListDevicesMap(ctx context.Context) (map[string]string, error) { return out, rows.Err() } +// RebindDevicesByMachineID moves sessions off every other devices row that +// shares d.MachineID onto d.ID, then deletes those rows. d must already be +// stored: sessions.device_id references devices.id, and that column is the +// only local reference to a device id. Returns the number of sessions +// rebound. An empty MachineID matches nothing. +func (s *Store) RebindDevicesByMachineID(ctx context.Context, d Device) (int64, error) { + if d.ID == "" || d.MachineID == "" { + return 0, nil + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return 0, err + } + defer func() { _ = tx.Rollback() }() + + res, err := tx.ExecContext(ctx, ` + UPDATE sessions + SET device_id = ? + WHERE device_id IN ( + SELECT id FROM devices + WHERE machine_id = ? AND id != ? + ) + `, d.ID, d.MachineID, d.ID) + if err != nil { + return 0, fmt.Errorf("rebind sessions for machine %s: %w", d.MachineID, err) + } + n, _ := res.RowsAffected() + + if _, err := tx.ExecContext(ctx, ` + DELETE FROM devices + WHERE machine_id = ? AND id != ? + `, d.MachineID, d.ID); err != nil { + return 0, fmt.Errorf("delete stale devices for machine %s: %w", d.MachineID, err) + } + if err := tx.Commit(); err != nil { + return 0, err + } + return n, nil +} + // RebindLocalSessions reassigns every `device_id = 'local'` session row to // the given fingerprint, in one transaction. Returns the count rewritten. // Used during startup to migrate sessions imported under the seed device id diff --git a/internal/store/devices_test.go b/internal/store/devices_test.go index 3b0fa5ac..2e19eb19 100644 --- a/internal/store/devices_test.go +++ b/internal/store/devices_test.go @@ -119,3 +119,131 @@ func TestRebindLocalSessionsRejectsSelfId(t *testing.T) { require.NoError(t, err) require.Equal(t, int64(0), n) } + +func TestRebindDevicesByMachineIDCollapsesSameMachine(t *testing.T) { + t.Parallel() + ctx, s := newStore(t) + now := time.Now().UTC() + + canonical := Device{ + ID: "canon", Hostname: "tbox", MachineID: "uuid-1", + FriendlyName: "tbox", FingerprintedAt: now, + } + staleA := Device{ + ID: "stale-a", Hostname: "192.168.0.19", MachineID: "uuid-1", + FriendlyName: "192.168.0.19", FingerprintedAt: now.Add(-time.Hour), + } + staleB := Device{ + ID: "stale-b", Hostname: "ip-192-168-0-16", MachineID: "uuid-1", + FriendlyName: "ip-192-168-0-16", FingerprintedAt: now.Add(-2 * time.Hour), + } + // Same machine, no sessions: the row is still removed. + staleC := Device{ + ID: "stale-c", Hostname: "10.0.0.8", MachineID: "uuid-1", + FriendlyName: "10.0.0.8", FingerprintedAt: now.Add(-3 * time.Hour), + } + foreign := Device{ + ID: "other", Hostname: "laptop", MachineID: "uuid-2", + FriendlyName: "laptop", FingerprintedAt: now, + } + blank := Device{ + ID: "blank", Hostname: "noname", MachineID: "", + FriendlyName: "noname", FingerprintedAt: now, + } + for _, d := range []Device{canonical, staleA, staleB, staleC, foreign, blank} { + require.NoError(t, s.UpsertDevice(ctx, d)) + } + + seed := func(id, deviceID string) { + t.Helper() + sess := newSession(id, now) + sess.DeviceID = deviceID + require.NoError(t, s.UpsertSession(ctx, sess, nil)) + } + // Two on the first stale id, one on the second, one already canonical, + // one on a different machine, one on an empty machine id. + seed("s1", "stale-a") + seed("s2", "stale-a") + seed("s3", "stale-b") + seed("s4", "canon") + seed("s5", "other") + seed("s6", "blank") + + n, err := s.RebindDevicesByMachineID(ctx, canonical) + require.NoError(t, err) + require.Equal(t, int64(3), n) + + var total int + require.NoError(t, s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM sessions`).Scan(&total)) + require.Equal(t, 6, total) + + wantDevice := map[string]string{ + "s1": "canon", + "s2": "canon", + "s3": "canon", + "s4": "canon", + "s5": "other", + "s6": "blank", + } + for id, want := range wantDevice { + var got string + require.NoError(t, s.DB().QueryRowContext(ctx, + `SELECT device_id FROM sessions WHERE id = ?`, id).Scan(&got)) + require.Equal(t, want, got, id) + } + + devs, err := s.ListDevices(ctx) + require.NoError(t, err) + gotIDs := map[string]struct{}{} + for _, d := range devs { + gotIDs[d.ID] = struct{}{} + } + require.Contains(t, gotIDs, "canon") + require.Contains(t, gotIDs, "other") + require.Contains(t, gotIDs, "blank") + require.Contains(t, gotIDs, "local") + require.NotContains(t, gotIDs, "stale-a") + require.NotContains(t, gotIDs, "stale-b") + require.NotContains(t, gotIDs, "stale-c") + + n, err = s.RebindDevicesByMachineID(ctx, canonical) + require.NoError(t, err) + require.Equal(t, int64(0), n) + + devs, err = s.ListDevices(ctx) + require.NoError(t, err) + require.Len(t, devs, 4) +} + +func TestRebindDevicesByMachineIDIgnoresEmptyMachineID(t *testing.T) { + t.Parallel() + ctx, s := newStore(t) + now := time.Now().UTC() + + a := Device{ID: "a", Hostname: "a", MachineID: "", FriendlyName: "a", FingerprintedAt: now} + b := Device{ID: "b", Hostname: "b", MachineID: "", FriendlyName: "b", FingerprintedAt: now} + require.NoError(t, s.UpsertDevice(ctx, a)) + require.NoError(t, s.UpsertDevice(ctx, b)) + + sess := newSession("s", now) + sess.DeviceID = "b" + require.NoError(t, s.UpsertSession(ctx, sess, nil)) + + n, err := s.RebindDevicesByMachineID(ctx, a) + require.NoError(t, err) + require.Equal(t, int64(0), n) + + var got string + require.NoError(t, s.DB().QueryRowContext(ctx, + `SELECT device_id FROM sessions WHERE id = 's'`).Scan(&got)) + require.Equal(t, "b", got) + + devs, err := s.ListDevices(ctx) + require.NoError(t, err) + ids := map[string]struct{}{} + for _, d := range devs { + ids[d.ID] = struct{}{} + } + require.Contains(t, ids, "a") + require.Contains(t, ids, "b") +} diff --git a/internal/store/prune.go b/internal/store/prune.go index 2ab7633e..dd72d4cf 100644 --- a/internal/store/prune.go +++ b/internal/store/prune.go @@ -114,6 +114,26 @@ func (s *Store) ListPruneCandidates(ctx context.Context, deviceID string, before return out, rows.Err() } +// CountOldUnconfirmed counts sessions on deviceID that are inactive before +// the cutoff, not pruned, and not confirmed: pushed_hash is NULL or differs +// from raw_hash. ListPruneCandidates leaves these rows out. +func (s *Store) CountOldUnconfirmed(ctx context.Context, deviceID string, before time.Time) (int, error) { + var n int + err := s.db.QueryRowContext(ctx, ` + SELECT COUNT(*) + FROM sessions s + JOIN sync_state ss ON ss.session_id = s.id + WHERE s.device_id = ? + AND s.pruned_at IS NULL + AND s.last_activity_at < ? + AND (ss.pushed_hash IS NULL OR ss.pushed_hash != s.raw_hash) + `, deviceID, formatTime(before)).Scan(&n) + if err != nil { + return 0, fmt.Errorf("count unconfirmed sessions: %w", err) + } + return n, nil +} + // MarkPruned flips a session to pruned, guarded by the raw hash observed at // candidate time and by not being pruned already. Returns false when the // guard failed (raw changed or a concurrent prune won) — the caller must diff --git a/internal/store/prune_test.go b/internal/store/prune_test.go index 7494e183..c35b4eb6 100644 --- a/internal/store/prune_test.go +++ b/internal/store/prune_test.go @@ -120,6 +120,96 @@ func seedPrunable(t *testing.T, s *Store, id string, ago time.Duration) session. return sess } +func TestListPruneCandidatesAfterDeviceRebind(t *testing.T) { + t.Parallel() + ctx, s := newStore(t) + now := time.Now().UTC() + + canonical := Device{ + ID: "canon", Hostname: "tbox", MachineID: "uuid-1", + FriendlyName: "tbox", FingerprintedAt: now, + } + stale := Device{ + ID: "stale", Hostname: "192.168.0.19", MachineID: "uuid-1", + FriendlyName: "192.168.0.19", FingerprintedAt: now.Add(-time.Hour), + } + require.NoError(t, s.UpsertDevice(ctx, canonical)) + require.NoError(t, s.UpsertDevice(ctx, stale)) + + sess := newSession("old-stale", now) + sess.DeviceID = stale.ID + sess.RawHash = "abc123hash" + sess.StartedAt = now.Add(-60 * 24 * time.Hour) + sess.LastActivityAt = now.Add(-40 * 24 * time.Hour) + require.NoError(t, s.UpsertSession(ctx, sess, nil)) + require.NoError(t, s.RecordSync(ctx, sess.ID, sess.RawHash)) + require.NoError(t, s.RecordPushed(ctx, sess.ID, sess.RawHash, "s3://bucket/"+sess.ID)) + + before := now.Add(-30 * 24 * time.Hour) + got, err := s.ListPruneCandidates(ctx, canonical.ID, before, 0) + require.NoError(t, err) + require.Empty(t, got) + + n, err := s.RebindDevicesByMachineID(ctx, canonical) + require.NoError(t, err) + require.Equal(t, int64(1), n) + + got, err = s.ListPruneCandidates(ctx, canonical.ID, before, 0) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, sess.ID, got[0].ID) + require.Equal(t, sess.RawHash, got[0].RawHash) + require.Equal(t, sess.Agent, got[0].Agent) +} + +func TestCountOldUnconfirmed(t *testing.T) { + t.Parallel() + ctx, s := newStore(t) + now := time.Now().UTC() + before := now.Add(-30 * 24 * time.Hour) + + oldNull := newSession("old-null", now) + oldNull.LastActivityAt = now.Add(-40 * 24 * time.Hour) + require.NoError(t, s.UpsertSession(ctx, oldNull, nil)) + require.NoError(t, s.RecordSync(ctx, oldNull.ID, oldNull.RawHash)) + + diverged := seedPrunable(t, s, "old-diverged", 40*24*time.Hour) + require.NoError(t, s.RecordPushed(ctx, diverged.ID, "stale-hash", "")) + + seedPrunable(t, s, "old-confirmed", 40*24*time.Hour) + + recentNull := newSession("recent-null", now) + recentNull.LastActivityAt = now.Add(-time.Hour) + require.NoError(t, s.UpsertSession(ctx, recentNull, nil)) + require.NoError(t, s.RecordSync(ctx, recentNull.ID, recentNull.RawHash)) + + pruned := newSession("old-pruned", now) + pruned.LastActivityAt = now.Add(-40 * 24 * time.Hour) + require.NoError(t, s.UpsertSession(ctx, pruned, nil)) + require.NoError(t, s.RecordSync(ctx, pruned.ID, pruned.RawHash)) + ok, err := s.MarkPruned(ctx, pruned.ID, pruned.RawHash) + require.NoError(t, err) + require.True(t, ok) + + require.NoError(t, s.UpsertDevice(ctx, Device{ + ID: "other", Hostname: "laptop", MachineID: "mid-other", + FriendlyName: "laptop", FingerprintedAt: now, + })) + foreign := newSession("old-foreign", now) + foreign.DeviceID = "other" + foreign.LastActivityAt = now.Add(-40 * 24 * time.Hour) + require.NoError(t, s.UpsertSession(ctx, foreign, nil)) + require.NoError(t, s.RecordSync(ctx, foreign.ID, foreign.RawHash)) + + noState := newSession("old-no-sync-state", now) + noState.LastActivityAt = now.Add(-40 * 24 * time.Hour) + require.NoError(t, s.UpsertSession(ctx, noState, nil)) + + n, err := s.CountOldUnconfirmed(ctx, "local", before) + require.NoError(t, err) + require.Equal(t, 2, n) +} + func TestListPruneCandidatesFilters(t *testing.T) { t.Parallel() ctx, s := newStore(t)