diff --git a/native/macos/CuedNative/Sources/CuedNative/RootWindowController.swift b/native/macos/CuedNative/Sources/CuedNative/RootWindowController.swift index e7152660..795dfbcf 100644 --- a/native/macos/CuedNative/Sources/CuedNative/RootWindowController.swift +++ b/native/macos/CuedNative/Sources/CuedNative/RootWindowController.swift @@ -203,11 +203,6 @@ final class RootWindowController: NSWindowController { return } - let globalSkill = Self.decodeJSON( - daemonSupervisor: daemonSupervisor, - SkillStatus.self, - arguments: ["skill", "status"] - ) ?? snapshot.globalSkill let setupIntegrations = snapshot.setupIntegrations.filter { $0.capability.setupVisible } @@ -222,7 +217,7 @@ final class RootWindowController: NSWindowController { self.isRefreshing = false self.viewModel.apply( permissions: permissions, - globalSkill: globalSkill, + globalSkill: snapshot.globalSkill, relationshipSummary: snapshot.relationshipSummary ?? RelationshipSummary(), allIntegrations: self.cachedAllIntegrations, integrations: setupIntegrations diff --git a/src/cli.ts b/src/cli.ts index 3cd03997..d59d59ea 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -58,6 +58,7 @@ import { import { readMenuBarStatusCache } from "./runtime/menu-bar-status-cache.js"; import { buildOnboardingSnapshot } from "./runtime/onboarding.js"; import { runProjectionWorkerFromEnv } from "./runtime/projection/worker.js"; +import { hydrateIntegrationProjectionCounts } from "./runtime/status.js"; import { checkForUpdates, clearUpdateHelperPendingState, @@ -584,7 +585,10 @@ async function main(): Promise { lastUpdatedAt: Date.now(), source: "menu_bar_cache", }, - integrations: listMenuBarIntegrationStates(db), + integrations: hydrateIntegrationProjectionCounts( + listMenuBarIntegrationStates(db), + overview.integrationBreakdown, + ), update: getUpdateStatus(db), socketRunning: existsSync(CUED_SOCKET_PATH), dbPath: db.dbPath, diff --git a/src/db/database.test.ts b/src/db/database.test.ts index 1d19a1aa..b1824954 100644 --- a/src/db/database.test.ts +++ b/src/db/database.test.ts @@ -70,6 +70,114 @@ describe("CuedDatabase", () => { db.close(); }); + it("uses the normalized schema expression index for raw event diagnostics", () => { + const db = createDb(); + const sqlite = ( + db as unknown as { + sqlite: { + prepare: (sql: string) => { + all: (...params: unknown[]) => Array<{ detail: string }>; + }; + }; + } + ).sqlite; + const phonePlan = sqlite + .prepare( + ` + EXPLAIN QUERY PLAN + SELECT COUNT(*) + FROM raw_events + WHERE platform = ? + AND account_key = ? + AND COALESCE(normalized_schema, entity_kind || '.' || event_kind || '@1') = ? + `, + ) + .all("imessage", "local", "call.observed@1"); + const schemaPlan = sqlite + .prepare( + ` + EXPLAIN QUERY PLAN + SELECT + COALESCE(normalized_schema, entity_kind || '.' || event_kind || '@1') AS schema_key, + COUNT(*) + FROM raw_events + WHERE platform = ? AND account_key = ? + GROUP BY schema_key + `, + ) + .all("imessage", "local"); + + expect(phonePlan.map((row) => row.detail).join("\n")).toContain( + "idx_raw_events_platform_account_normalized_schema", + ); + expect(schemaPlan.map((row) => row.detail).join("\n")).toContain( + "idx_raw_events_platform_account_normalized_schema", + ); + expect(schemaPlan.map((row) => row.detail).join("\n")).not.toContain( + "USE TEMP B-TREE FOR GROUP BY", + ); + + db.close(); + }); + + it("groups menu bar projection counts by integration", () => { + const db = createDb(); + const sqlite = ( + db as unknown as { + sqlite: { exec: (sql: string) => void }; + } + ).sqlite; + sqlite.exec(` + INSERT INTO contacts (id, kind, name, archived, created_at, updated_at) + VALUES ('contact-1', 'person', 'Ava', 0, 1, 1); + INSERT INTO contact_sources ( + id, contact_id, platform, account_key, source_entity_key, first_seen_at, last_seen_at + ) VALUES ('source-1', 'contact-1', 'imessage', 'local', 'ava', 1, 1); + INSERT INTO conversations ( + id, platform, account_key, source_conversation_key, type, is_active, + unread_count, created_at, updated_at + ) VALUES + ('conversation-1', 'imessage', 'local', 'chat-1', 'dm', 1, 0, 1, 1), + ('conversation-2', 'slack', 'workspace', 'channel-1', 'channel', 1, 0, 1, 1); + INSERT INTO messages ( + id, platform, account_key, platform_message_id, conversation_id, sent_at, + is_from_me, is_deleted, is_edited, attachment_count, reaction_count, created_at, updated_at + ) VALUES + ('message-1', 'imessage', 'local', 'message-1', 'conversation-1', 1, 0, 0, 0, 0, 0, 1, 1), + ('message-2', 'slack', 'workspace', 'message-2', 'conversation-2', 1, 0, 0, 0, 0, 0, 1, 1); + `); + + expect(db.listProjectionCountsByIntegration()).toEqual( + expect.arrayContaining([ + { platform: "imessage", accountKey: "local", contacts: 1, messages: 1 }, + { platform: "slack", accountKey: "workspace", contacts: 0, messages: 1 }, + ]), + ); + sqlite.exec(` + INSERT INTO messages ( + id, platform, account_key, platform_message_id, conversation_id, sent_at, + is_from_me, is_deleted, is_edited, attachment_count, reaction_count, created_at, updated_at + ) VALUES ( + 'message-3', 'imessage', 'local', 'message-3', 'conversation-1', 2, + 0, 0, 0, 0, 0, 2, 2 + ); + `); + expect( + db + .listProjectionCountsByIntegration() + .find((row) => row.platform === "imessage" && row.accountKey === "local")?.messages, + ).toBe(1); + + db.upsertProjectionState({ projectionWatermark: 1 }); + expect( + db + .listProjectionCountsByIntegration() + .find((row) => row.platform === "imessage" && row.accountKey === "local")?.messages, + ).toBe(2); + + db.close(); + }); + function insertContact( db: CuedDatabase, input: { diff --git a/src/db/database.ts b/src/db/database.ts index b9e6afac..c4cc5c95 100644 --- a/src/db/database.ts +++ b/src/db/database.ts @@ -446,6 +446,13 @@ type WhatsAppSendCandidate = { resolution: Extract; }; +export interface IntegrationProjectionCount { + platform: Platform; + accountKey: string; + contacts: number; + messages: number; +} + const SIGNAL_UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; @@ -581,6 +588,11 @@ function buildRawEventValues(event: RawEventInput) { export class CuedDatabase { private readonly sqlite: Database.Database; private readonly db: LocalDrizzleDatabase; + private menuBarProjectionCountsCache: { + projectionWatermark: number; + lastRebuildAt: number | null; + rows: IntegrationProjectionCount[]; + } | null = null; constructor( public readonly dbPath: string = CUED_DB_PATH, @@ -1416,8 +1428,22 @@ export class CuedDatabase { platform: Platform; messages: number; }>; + integrationBreakdown: IntegrationProjectionCount[]; } { - const messageBreakdown = this.listMessageCountsByPlatform(); + const integrationBreakdown = this.listProjectionCountsByIntegration(); + const messagesByPlatform = new Map(); + for (const row of integrationBreakdown) { + messagesByPlatform.set( + row.platform, + (messagesByPlatform.get(row.platform) ?? 0) + row.messages, + ); + } + const messageBreakdown = [...messagesByPlatform] + .map(([platform, messages]) => ({ platform, messages })) + .sort( + (left, right) => + right.messages - left.messages || left.platform.localeCompare(right.platform), + ); return { contacts: this.countRows(contacts), conversations: this.countRows(conversations), @@ -1427,9 +1453,68 @@ export class CuedDatabase { integrations: this.countRows(integrationStates), authSessions: this.countRows(authSessions), messageBreakdown, + integrationBreakdown, }; } + listProjectionCountsByIntegration(): IntegrationProjectionCount[] { + const projectionState = this.getProjectionState({ initialize: false }); + if ( + this.menuBarProjectionCountsCache?.projectionWatermark === + projectionState.projection_watermark && + this.menuBarProjectionCountsCache.lastRebuildAt === projectionState.last_rebuild_at + ) { + return this.menuBarProjectionCountsCache.rows; + } + + const byIntegration = new Map(); + const getOrCreate = (platform: Platform, accountKey: string) => { + const key = `${platform}\u0000${accountKey}`; + const existing = byIntegration.get(key); + if (existing) { + return existing; + } + const created = { platform, accountKey, contacts: 0, messages: 0 }; + byIntegration.set(key, created); + return created; + }; + const contactCounts = this.sqlite + .prepare( + ` + SELECT platform, account_key, COUNT(*) AS count + FROM contact_sources + GROUP BY platform, account_key + `, + ) + .all() as Array<{ platform: Platform; account_key: string; count: number | null }>; + for (const row of contactCounts) { + getOrCreate(row.platform, row.account_key).contacts = Number(row.count ?? 0); + } + const messageCounts = this.sqlite + .prepare( + ` + SELECT platform, account_key, COUNT(*) AS count + FROM messages + GROUP BY platform, account_key + `, + ) + .all() as Array<{ platform: Platform; account_key: string; count: number | null }>; + for (const row of messageCounts) { + getOrCreate(row.platform, row.account_key).messages = Number(row.count ?? 0); + } + const rows = [...byIntegration.values()]; + this.menuBarProjectionCountsCache = { + projectionWatermark: projectionState.projection_watermark, + lastRebuildAt: projectionState.last_rebuild_at, + rows, + }; + return rows; + } + + invalidateMenuBarProjectionCounts(): void { + this.menuBarProjectionCountsCache = null; + } + listMessageCountsByPlatform(): Array<{ platform: Platform; messages: number; @@ -1602,7 +1687,7 @@ export class CuedDatabase { .prepare( ` SELECT - COALESCE(normalized_schema, entity_kind || '.' || event_kind) AS schema_key, + COALESCE(normalized_schema, entity_kind || '.' || event_kind || '@1') AS schema_key, COUNT(*) AS count FROM raw_events WHERE platform = ? AND account_key = ? @@ -4908,6 +4993,7 @@ export class CuedDatabase { tx.delete(contactSources).run(); tx.delete(contacts).run(); }); + this.invalidateMenuBarProjectionCounts(); } private countRows(table: SQLiteTable): number { diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 0a14ba55..388965c0 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -2229,4 +2229,20 @@ export const MIGRATIONS: Migration[] = [ `); }, }, + { + id: "0021_raw_event_normalized_schema_index", + apply: (db) => { + if (!tableExists(db, "raw_events")) { + return; + } + db.exec(` + CREATE INDEX IF NOT EXISTS idx_raw_events_platform_account_normalized_schema + ON raw_events( + platform, + account_key, + COALESCE(normalized_schema, entity_kind || '.' || event_kind || '@1') + ); + `); + }, + }, ]; diff --git a/src/platforms/core/state/status.ts b/src/platforms/core/state/status.ts index b475f640..f54668ff 100644 --- a/src/platforms/core/state/status.ts +++ b/src/platforms/core/state/status.ts @@ -131,6 +131,11 @@ const AUTH_SESSION_STALE_AFTER_MS = 15 * 60 * 1000; const AUTH_SESSION_ABANDONED_ERROR = "Auth session ended before Cued received a completion event"; const AUTH_SESSION_EXPIRED_ERROR = "Auth session expired before completion"; +interface IntegrationStatusOptions { + includeLiveLocalIntegrations?: boolean; + includeDiagnostics?: boolean; +} + export function now(): number { return Date.now(); } @@ -687,7 +692,8 @@ function compareSetupIntegrations( function buildSetupIntegrations( db: CuedDatabase, - options: { includeLiveLocalIntegrations?: boolean; includeDiagnostics?: boolean } = {}, + options: IntegrationStatusOptions = {}, + existingIntegrations?: IntegrationStateSummary[], ): IntegrationStateSummary[] { const onboardingOrder: Platform[] = [ "contacts", @@ -700,7 +706,7 @@ function buildSetupIntegrations( "signal", ]; const byPlatform = new Map(); - for (const integration of listIntegrationStates(db, options)) { + for (const integration of existingIntegrations ?? listIntegrationStates(db, options)) { if (!isOnboardingVisiblePlatform(integration.platform)) { continue; } @@ -895,15 +901,16 @@ export function getPlatformRuntimeDefaults(platform: Platform): { export function buildIntegrationStatus( db: CuedDatabase, - options: { includeLiveLocalIntegrations?: boolean; includeDiagnostics?: boolean } = {}, + options: IntegrationStatusOptions = {}, ): { hostOs: ReturnType; integrations: IntegrationStateSummary[]; setupIntegrations: IntegrationStateSummary[]; } { + const integrations = listIntegrationStates(db, options); return { hostOs: resolveHostOS(), - integrations: listIntegrationStates(db, options), - setupIntegrations: buildSetupIntegrations(db, options), + integrations, + setupIntegrations: buildSetupIntegrations(db, options, integrations), }; } diff --git a/src/runtime/onboarding-permissions.test.ts b/src/runtime/onboarding-permissions.test.ts index c5db368c..ae81cbdf 100644 --- a/src/runtime/onboarding-permissions.test.ts +++ b/src/runtime/onboarding-permissions.test.ts @@ -38,7 +38,7 @@ describe("onboarding permission refresh", () => { dbPath: "/tmp/cued-test/local.db", getMenuBarOverview: () => ({ messages: 0, contacts: 0 }), getProjectionBacklog: () => ({ pending_raw_events: 0 }), - getIntegrationProjectionStats: () => ({ rawEventsBySchema: {} }), + countObservedPhoneCalls: () => 0, getAppSetting: vi.fn().mockReturnValue(null), }; @@ -201,4 +201,66 @@ describe("onboarding permission refresh", () => { source: "menu_bar_cache", }); }); + + it("rebuilds integration counts when a legacy cache has no breakdown", async () => { + mockSnapshotDependencies(); + buildIntegrationStatusMock.mockReturnValue({ + hostOs: "macos", + integrations: [ + { + platform: "imessage", + accountKey: "local", + projectionStats: { + rawEvents: 0, + rawEventsBySchema: {}, + projectedContacts: 0, + projectedConversations: 0, + projectedMessages: 0, + }, + }, + ], + setupIntegrations: [], + }); + readMenuBarStatusCacheMock.mockReturnValue({ + path: "/tmp/cued-test/menu-bar-status.json", + mtimeMs: 6789, + snapshot: { + dbPath: db.dbPath, + relationshipSummary: { + messages: 12, + contacts: 7, + phoneCalls: 3, + pendingProjectionEvents: 0, + }, + overview: { + messages: 12, + contacts: 7, + }, + }, + }); + const getMenuBarOverview = vi.fn().mockReturnValue({ + messages: 12, + contacts: 7, + integrationBreakdown: [ + { + platform: "imessage", + accountKey: "local", + contacts: 4, + messages: 10, + }, + ], + }); + + const snapshot = await buildOnboardingSnapshot({ ...db, getMenuBarOverview } as never, { + installGlobalSkill: false, + }); + + expect(getMenuBarOverview).toHaveBeenCalledTimes(1); + expect(snapshot.integrations[0]?.projectionStats).toEqual( + expect.objectContaining({ + projectedContacts: 4, + projectedMessages: 10, + }), + ); + }); }); diff --git a/src/runtime/onboarding.test.ts b/src/runtime/onboarding.test.ts index 1fa6faec..4ba9ca17 100644 --- a/src/runtime/onboarding.test.ts +++ b/src/runtime/onboarding.test.ts @@ -1,7 +1,7 @@ import { chmodSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { CuedDatabase } from "../db/database.js"; import { buildOnboardingSnapshot } from "./onboarding.js"; @@ -198,4 +198,68 @@ describe("onboarding snapshot", () => { db.close(); }); + + it("does not load raw schema diagnostics for onboarding", async () => { + const db = createDb(); + const projectionStats = vi.spyOn(db, "getIntegrationProjectionStats"); + + await buildOnboardingSnapshot(db); + + expect(projectionStats).not.toHaveBeenCalled(); + + db.close(); + }); + + it("builds platform breakdowns directly when the menu bar cache is unavailable", async () => { + const db = createDb(); + db.upsertIntegrationState({ + platform: "imessage", + accountKey: "local", + displayName: "Messages", + authState: "authorized", + enabled: true, + connectionKind: "native", + syncCapable: false, + launchStrategy: "system-settings", + launchTarget: null, + importedFrom: "test", + }); + const sqlite = ( + db as unknown as { + sqlite: { exec: (sql: string) => void }; + } + ).sqlite; + sqlite.exec(` + INSERT INTO contacts (id, kind, name, archived, created_at, updated_at) + VALUES ('contact-1', 'person', 'Ava', 0, 1, 1); + INSERT INTO contact_sources ( + id, contact_id, platform, account_key, source_entity_key, first_seen_at, last_seen_at + ) VALUES ('source-1', 'contact-1', 'imessage', 'local', 'ava', 1, 1); + INSERT INTO conversations ( + id, platform, account_key, source_conversation_key, type, is_active, + unread_count, created_at, updated_at + ) VALUES ('conversation-1', 'imessage', 'local', 'chat-1', 'dm', 1, 0, 1, 1); + INSERT INTO messages ( + id, platform, account_key, platform_message_id, conversation_id, sent_at, + is_from_me, is_deleted, is_edited, attachment_count, reaction_count, created_at, updated_at + ) VALUES ( + 'message-1', 'imessage', 'local', 'message-1', 'conversation-1', 1, + 0, 0, 0, 0, 0, 1, 1 + ); + `); + + const snapshot = await buildOnboardingSnapshot(db); + const imessage = snapshot.integrations.find( + (integration) => integration.platform === "imessage" && integration.accountKey === "local", + ); + + expect(imessage?.projectionStats).toEqual( + expect.objectContaining({ + projectedContacts: 1, + projectedMessages: 1, + }), + ); + + db.close(); + }); }); diff --git a/src/runtime/onboarding.ts b/src/runtime/onboarding.ts index f011de1d..d5725e5a 100644 --- a/src/runtime/onboarding.ts +++ b/src/runtime/onboarding.ts @@ -1,10 +1,12 @@ -import type { CuedDatabase } from "../db/database.js"; +import type { CuedDatabase, IntegrationProjectionCount } from "../db/database.js"; import { refreshManagedIntegrationStates } from "../platforms/core/state/refresh.js"; import { buildIntegrationStatus } from "../platforms/core/state/status.js"; +import { isPlatform } from "../platforms/core/types.js"; import { getGlobalCuedSkillStatus, installGlobalCuedSkill } from "../skills/install.js"; import { isTelemetryEnabled } from "../telemetry/client.js"; import { buildPermissionStatus } from "./doctor.js"; import { type MenuBarStatusCacheEntry, readMenuBarStatusCache } from "./menu-bar-status-cache.js"; +import { hydrateIntegrationProjectionCounts } from "./status.js"; export interface OnboardingSnapshot { permissions: Awaited>["permissions"]; @@ -65,55 +67,83 @@ export async function buildOnboardingSnapshot( }; } } - const integrations = buildIntegrationStatus(db, { includeDiagnostics: true }); + const cachedStatus = readMenuBarStatusCache({ dbPath: db.dbPath }); + const cachedRelationship = cachedStatus ? readCachedRelationshipSummary(cachedStatus) : null; + const cachedProjectionCounts = cachedStatus ? readCachedProjectionCounts(cachedStatus) : null; + const fallbackOverview = + cachedRelationship && cachedProjectionCounts ? null : db.getMenuBarOverview(); + const projectionCounts = cachedProjectionCounts ?? fallbackOverview?.integrationBreakdown ?? []; + const integrationStatus = buildIntegrationStatus(db, { includeDiagnostics: false }); + const integrations = hydrateIntegrationProjectionCounts( + integrationStatus.integrations, + projectionCounts, + ); + const setupIntegrations = hydrateIntegrationProjectionCounts( + integrationStatus.setupIntegrations, + projectionCounts, + ); return { permissions: permissions.permissions, globalSkill, - ...integrations, - relationshipSummary: buildRelationshipSummary(db), + ...integrationStatus, + integrations, + setupIntegrations, + relationshipSummary: + cachedRelationship ?? buildDatabaseRelationshipSummary(db, fallbackOverview), }; } -function buildRelationshipSummary(db: CuedDatabase): RelationshipSummary { - const cached = readCachedRelationshipSummary(db.dbPath); - if (cached) { - return { - ...cached, - phoneCalls: - cached.phoneCalls > 0 - ? cached.phoneCalls - : (db.getIntegrationProjectionStats("imessage", "local").rawEventsBySchema[ - "call.observed@1" - ] ?? 0), - }; - } - - const overview = db.getMenuBarOverview(); +function buildDatabaseRelationshipSummary( + db: CuedDatabase, + overview: ReturnType | null, +): RelationshipSummary { + const resolvedOverview = overview ?? db.getMenuBarOverview(); const projection = db.getProjectionBacklog({ initializeProjectionState: false }); - const imessageStats = db.getIntegrationProjectionStats("imessage", "local"); return { - messages: overview.messages, - contacts: overview.contacts, - phoneCalls: imessageStats.rawEventsBySchema["call.observed@1"] ?? 0, + messages: resolvedOverview.messages, + contacts: resolvedOverview.contacts, + phoneCalls: db.countObservedPhoneCalls(), pendingProjectionEvents: projection.pending_raw_events, lastUpdatedAt: null, source: "database", }; } -function readCachedRelationshipSummary(dbPath: string): RelationshipSummary | null { - const cached = readMenuBarStatusCache({ dbPath }); - if (!cached) { - return null; - } - +function readCachedRelationshipSummary( + cached: MenuBarStatusCacheEntry, +): RelationshipSummary | null { return ( readCurrentCachedRelationshipSummary(cached) ?? readLegacyCachedRelationshipSummary(cached) ); } +function readCachedProjectionCounts( + cached: MenuBarStatusCacheEntry, +): IntegrationProjectionCount[] | null { + const overview = isRecord(cached.snapshot.overview) ? cached.snapshot.overview : null; + if (!Array.isArray(overview?.integrationBreakdown)) { + return null; + } + const counts: IntegrationProjectionCount[] = []; + for (const row of overview.integrationBreakdown) { + if (!isRecord(row) || typeof row.platform !== "string" || !isPlatform(row.platform)) { + return null; + } + if (typeof row.accountKey !== "string") { + return null; + } + counts.push({ + platform: row.platform, + accountKey: row.accountKey, + contacts: numberValue(row.contacts), + messages: numberValue(row.messages), + }); + } + return counts; +} + function readCurrentCachedRelationshipSummary( cached: MenuBarStatusCacheEntry, ): RelationshipSummary | null { diff --git a/src/runtime/projection/projector.test.ts b/src/runtime/projection/projector.test.ts index f90ae4af..4ad0fa68 100644 --- a/src/runtime/projection/projector.test.ts +++ b/src/runtime/projection/projector.test.ts @@ -1938,6 +1938,79 @@ describe("projector", () => { db.close(); }); + it("invalidates cached integration counts after out-of-order realtime projection", () => { + const db = createDb(); + db.insertRawEvent({ + id: "conversation-count-cache", + platform: "linkedin", + accountKey: "default", + entityKind: "conversation", + eventKind: "observed", + observedAt: 1, + dedupeKey: "conversation-count-cache", + payload: { + sourceConversationKey: "thread-count-cache", + conversationType: "dm", + participants: [], + }, + sourceVersion: "test-v1", + }); + projectPendingRawEvents(db); + + const outOfOrder = db.insertRawEvents([ + { + id: "contact-count-cache-filler", + platform: "contacts", + accountKey: "local", + entityKind: "contact", + eventKind: "observed", + observedAt: 2, + dedupeKey: "contact-count-cache-filler", + payload: { + sourceEntityKey: "contacts:filler", + fields: { display_name: "Filler" }, + handles: [], + }, + sourceVersion: "test-v1", + }, + { + id: "message-count-cache", + platform: "linkedin", + accountKey: "default", + entityKind: "message", + eventKind: "created", + observedAt: 3, + dedupeKey: "message-count-cache", + payload: { + sourceMessageKey: "message-count-cache", + sourceConversationKey: "thread-count-cache", + sentAt: 3, + content: "invalidate the count cache", + isFromMe: false, + }, + sourceVersion: "test-v1", + }, + ]); + const messageRowId = outOfOrder.lastInsertedRowId!; + const before = db + .listProjectionCountsByIntegration() + .find((row) => row.platform === "linkedin" && row.accountKey === "default"); + expect(before?.messages ?? 0).toBe(0); + + projectRealtimeRange(db, { + startRowId: messageRowId, + endRowId: messageRowId, + }); + + const after = db + .listProjectionCountsByIntegration() + .find((row) => row.platform === "linkedin" && row.accountKey === "default"); + expect(after?.messages).toBe(1); + expect(db.getProjectionState().projection_watermark).toBe(1); + + db.close(); + }); + it("normalizes attachment-only placeholders on realtime projection", () => { const db = createDb(); diff --git a/src/runtime/projection/projector.ts b/src/runtime/projection/projector.ts index 26d4c923..3b0e1a80 100644 --- a/src/runtime/projection/projector.ts +++ b/src/runtime/projection/projector.ts @@ -2430,6 +2430,7 @@ function projectEventBatch( }); } }); + db.invalidateMenuBarProjectionCounts(); for (const failure of projectionFailures) { db.quarantineRawEventProjectionFailure( { diff --git a/src/runtime/status.test.ts b/src/runtime/status.test.ts new file mode 100644 index 00000000..90b4c3c8 --- /dev/null +++ b/src/runtime/status.test.ts @@ -0,0 +1,89 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { CuedDatabase } from "../db/database.js"; +import { buildMenuBarDaemonStatusSnapshot } from "./status.js"; + +describe("menu bar status snapshot", () => { + const tempDirs: string[] = []; + + afterEach(() => { + while (tempDirs.length > 0) { + rmSync(tempDirs.pop()!, { recursive: true, force: true }); + } + }); + + it("includes cached message and contact counts for each integration", () => { + const dir = mkdtempSync(join(tmpdir(), "cued-menu-status-")); + tempDirs.push(dir); + const db = new CuedDatabase(join(dir, "local.db")); + db.migrate(); + db.upsertIntegrationState({ + platform: "imessage", + accountKey: "local", + displayName: "Messages", + authState: "authorized", + enabled: true, + connectionKind: "native", + syncCapable: false, + launchStrategy: "system-settings", + launchTarget: null, + importedFrom: "test", + }); + const sqlite = ( + db as unknown as { + sqlite: { exec: (sql: string) => void }; + } + ).sqlite; + sqlite.exec(` + INSERT INTO contacts (id, kind, name, archived, created_at, updated_at) + VALUES ('contact-1', 'person', 'Ava', 0, 1, 1); + INSERT INTO contact_sources ( + id, contact_id, platform, account_key, source_entity_key, first_seen_at, last_seen_at + ) VALUES ('source-1', 'contact-1', 'imessage', 'local', 'ava', 1, 1); + INSERT INTO conversations ( + id, platform, account_key, source_conversation_key, type, is_active, + unread_count, created_at, updated_at + ) VALUES ('conversation-1', 'imessage', 'local', 'chat-1', 'dm', 1, 0, 1, 1); + INSERT INTO messages ( + id, platform, account_key, platform_message_id, conversation_id, sent_at, + is_from_me, is_deleted, is_edited, attachment_count, reaction_count, created_at, updated_at + ) VALUES ( + 'message-1', 'imessage', 'local', 'message-1', 'conversation-1', 1, + 0, 0, 0, 0, 0, 1, 1 + ); + `); + const realtime = { getStatuses: () => [] }; + + const snapshot = buildMenuBarDaemonStatusSnapshot(db, { + app: {}, + discordRealtime: realtime as never, + slackRealtime: realtime as never, + linkedInRealtime: realtime as never, + signalRealtime: realtime as never, + whatsAppRealtime: realtime as never, + socketPath: join(dir, "daemon.sock"), + bootstrap: { + state: "ready", + startedAt: 1, + finishedAt: 2, + error: null, + }, + }); + + expect( + snapshot.integrations.find( + (integration) => integration.platform === "imessage" && integration.accountKey === "local", + )?.projectionStats, + ).toEqual( + expect.objectContaining({ + projectedContacts: 1, + projectedMessages: 1, + rawEventsBySchema: {}, + }), + ); + + db.close(); + }); +}); diff --git a/src/runtime/status.ts b/src/runtime/status.ts index 41567774..7b863c35 100644 --- a/src/runtime/status.ts +++ b/src/runtime/status.ts @@ -1,8 +1,9 @@ -import type { CuedDatabase } from "../db/database.js"; +import type { CuedDatabase, IntegrationProjectionCount } from "../db/database.js"; import { buildIntegrationStatus, listMenuBarIntegrationStates, } from "../platforms/core/state/status.js"; +import type { IntegrationStateSummary } from "../platforms/core/state/types.js"; import type { DiscordRealtimeSupervisor } from "../platforms/discord/realtime/session.js"; import type { LinkedInRealtimeSupervisor } from "../platforms/linkedin/realtime/session.js"; import type { SignalRealtimeSupervisor } from "../platforms/signal/realtime/session.js"; @@ -132,6 +133,10 @@ export function buildMenuBarDaemonStatusSnapshot( const projection = db.getProjectionBacklog({ initializeProjectionState: false }); const searchIndex = db.getMessageFtsIndexBacklog(); const phoneCalls = db.countObservedPhoneCalls(); + const integrations = hydrateIntegrationProjectionCounts( + listMenuBarIntegrationStates(db), + overview.integrationBreakdown, + ); return { app: options.app, bootstrap: options.bootstrap, @@ -155,7 +160,7 @@ export function buildMenuBarDaemonStatusSnapshot( lastUpdatedAt: Date.now(), source: "menu_bar_cache", }, - integrations: listMenuBarIntegrationStates(db), + integrations, discordRealtimeSessions: options.discordRealtime.getStatuses(), slackRealtimeSessions: options.slackRealtime.getStatuses(), linkedinRealtimeSessions: options.linkedInRealtime.getStatuses(), @@ -166,3 +171,25 @@ export function buildMenuBarDaemonStatusSnapshot( dbPath: db.dbPath, }; } + +export function hydrateIntegrationProjectionCounts( + integrations: IntegrationStateSummary[], + counts: IntegrationProjectionCount[], +): IntegrationStateSummary[] { + const countsByIntegration = new Map( + counts.map((row) => [`${row.platform}\u0000${row.accountKey}`, row]), + ); + return integrations.map((integration) => { + const projectionCounts = countsByIntegration.get( + `${integration.platform}\u0000${integration.accountKey}`, + ); + return { + ...integration, + projectionStats: { + ...integration.projectionStats, + projectedContacts: projectionCounts?.contacts ?? 0, + projectedMessages: projectionCounts?.messages ?? 0, + }, + }; + }); +}