diff --git a/.agents/skills/databuddy-internal/SKILL.md b/.agents/skills/databuddy-internal/SKILL.md index cfaa83462..71f6157fe 100644 --- a/.agents/skills/databuddy-internal/SKILL.md +++ b/.agents/skills/databuddy-internal/SKILL.md @@ -19,6 +19,7 @@ Keep additions **minimal**: one bullet, a new `rg` hint, or a routing note—eno ## Quick Map +- Before any PR merge, follow the AGENTS.md review-feedback gate: wait for configured reviewers on the final head, read all comment/review/thread pages, address each finding with evidence, and re-fetch to verify no unresolved feedback. Review bots can finish several minutes after a draft becomes ready; green CI does not establish completed review. - Prod infrastructure repo is local at `/Users/iza/Documents/GitHub/databuddy-infra` (`databuddy-analytics/infra`); ClickHouse cluster inventory is `clickhouse/ansible/inventory.yml`, not `/Users/iza/Dev/Databuddy/infra` or `DatabuddyOPS`. - Never use production/customer data as tests, fixtures, snapshots, examples, or copied output. Tests must use placeholders/mocks only (example.com, example IDs). If production ClickHouse is queried for investigation, summarize anonymized aggregates and do not paste customer domains, client IDs, emails, or other identifiers into code or responses. - `@databuddy/test/env` targets local `databuddy_test` unless `CI=true`, so a normal `db:push` may update a different database; sync that test database explicitly before debugging removed-column failures. diff --git a/AGENTS.md b/AGENTS.md index 0bec822a0..a66ef335d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -160,6 +160,7 @@ For picker controls, use the component that matches the interaction: - **Start fresh**: Check for an existing PR that owns the same surface, public contract, schema, or deployment configuration, then create the branch from an up-to-date `origin/staging`. Do not use an unmerged feature branch as a base unless the dependency is explicit, approved, and named as `Depends on #…` in both PRs. - **Make ownership visible**: Push and open a draft PR against `staging` once the slice has a first commit. State its scope, dependencies, and known overlaps. - **Keep integration linear**: Rebase a slice onto current `origin/staging` before it is ready for review; do not merge `staging` into the slice merely to refresh it. Request fresh review when a rebase changes reviewed code. +- **Resolve all review feedback before merging**: Mark the PR ready and wait for configured reviewers to finish on the final head; green CI alone is insufficient. Read every page of general comments, reviews, and inline threads, including outdated threads. Fix actionable findings or document a supported reason for declining them, then resolve each thread. Immediately before merging, re-fetch feedback and verify zero unresolved threads and no unaddressed comments or pending reviews. Never merge immediately after marking a draft ready or pushing review fixes while reviewers are still running. - **Isolate parallel work**: Use one worktree per active branch. Never let two agents or contributors mutate the same branch or reuse a task branch for a different concern. - **Retire completed work**: Merged PR source branches are automatically deleted. Delete closed PR branches manually, remove clean finished worktrees, and create a new branch from current `staging` for any follow-up—never revive or repurpose an old PR branch. diff --git a/apps/dashboard/app/(main)/organizations/components/measurement-plan-editor.tsx b/apps/dashboard/app/(main)/organizations/components/measurement-plan-editor.tsx index 9aaa74751..0edbaeb62 100644 --- a/apps/dashboard/app/(main)/organizations/components/measurement-plan-editor.tsx +++ b/apps/dashboard/app/(main)/organizations/components/measurement-plan-editor.tsx @@ -71,8 +71,8 @@ export function MeasurementPlanEditor({ return ( - Activation and return - + Activation and return + Choose the events that mean someone got value and came back. Saved definitions guide automatic investigations. Only identified profiles can be measured. @@ -88,7 +88,7 @@ export function MeasurementPlanEditor({ className="flex items-center justify-between gap-2 text-xs" key={item.websiteId} > -

+

{item.name || item.domain}: website unavailable. This definition is inactive.

@@ -120,25 +120,27 @@ export function MeasurementPlanEditor({ className="space-y-2 break-words text-xs" key={item.websiteId} > -

+

{item.name || "Unnamed outcome"}

-

{item.domain}

+

+ {item.domain} +

{site && site.domain !== item.domain && ( -

+

Website domain changed to {site.domain}. This definition is inactive until updated.

)} -

+

Activation: {item.activationEvent || "Not set"}

-

+

Return: {item.returnEvent || "Not set"} within{" "} {item.horizonDays} days

{item.namespace && ( -

+

Namespace: {item.namespace}

)} @@ -146,7 +148,7 @@ export function MeasurementPlanEditor({ ); }) ) : ( -

+

No definitions configured.

) @@ -182,7 +184,7 @@ export function MeasurementPlanEditor({ ) : ( -

+

{website.domain}

)} @@ -200,7 +202,7 @@ export function MeasurementPlanEditor({
{domainMismatch && (
-

+

This definition is bound to {plan.domain}. Update it to{" "} {website.domain} before saving.

@@ -233,7 +235,7 @@ export function MeasurementPlanEditor({ suggestions={events} value={plan[key]} /> - + {catalog.isError ? "Catalog unavailable; enter an exact name." : catalog.isPending @@ -293,7 +295,7 @@ export function MeasurementPlanEditor({
) : ( -

+

{plans.length >= 20 ? "Up to 20 website definitions are supported." : "No definition for this website. Add one to choose the outcome and events."} @@ -301,7 +303,7 @@ export function MeasurementPlanEditor({ )} ) : ( -

+

Add a website to define activation and return.

)} diff --git a/apps/insights/src/generation.ts b/apps/insights/src/generation.ts index ec884fd37..190497a85 100644 --- a/apps/insights/src/generation.ts +++ b/apps/insights/src/generation.ts @@ -108,6 +108,7 @@ import type { WebsiteInvestigation } from "./persistence"; import { isInterruptingInvestigation, persistInvestigation, + retireObsoleteRetentionObservation, } from "./persistence"; import { captureInsightsError, @@ -536,7 +537,7 @@ function annotationEvidence(rows: InvestigationAnnotation[]): string | null { return value.length <= 500 ? value : `${value.slice(0, 499).trimEnd()}…`; } -async function discoverWebsiteSignals( +export async function discoverWebsiteSignals( input: InvestigateWebsiteInput, runtime: InvestigationRuntime, options: { allowCoolingFallback?: boolean } = {} @@ -657,6 +658,17 @@ async function discoverWebsiteSignals( `Insight detection was incomplete (${metricDiagnostics.failedFamilies} metric families and ${definitionDiagnostics.failedDefinitions} conversion definitions failed)` ); } + const retiredDue = + due && + !remeasuredDue && + runtime.mode === "production" && + (await retireObsoleteRetentionObservation({ + asOf: asOf.toDate(), + domain: input.domain, + observation: due, + organizationId: input.organizationId, + websiteId: input.websiteId, + })); const signalsByKey = new Map(); for (const signal of [ ...(remeasuredDue ? [remeasuredDue] : []), @@ -670,12 +682,16 @@ async function discoverWebsiteSignals( signalsByKey.set(key, signal); } } + if (retiredDue && due) { + // A parallel detector may have read the definition before it was edited. + signalsByKey.delete(due.signal.signalKey); + } const detectedSignals = rankSignals([...signalsByKey.values()]); if (detectedSignals.length === 0) { const coverage = emptyInvestigationCoverage( - due ? "due_recheck_unmeasurable" : "no_detected_signals" + due && !retiredDue ? "due_recheck_unmeasurable" : "no_detected_signals" ); - if (due) { + if (due && !retiredDue) { if (runtime.mode === "production") { emitInsightsEvent( "info", @@ -735,7 +751,8 @@ async function discoverWebsiteSignals( : candidateAutomaticEligibleSignals; const hasDetectedCandidate = detectedSignals.some(isInvestigationCandidate); const hasPlannableCandidate = eligibleSignals.length > 0; - const hasUnmeasuredDue = due !== null && remeasuredDue === null; + const hasUnmeasuredDue = + due !== null && remeasuredDue === null && !retiredDue; if ( (hasUnmeasuredDue && !hasPlannableCandidate) || (eligibleSignals.length === 0 && !options.allowCoolingFallback) diff --git a/apps/insights/src/observations.ts b/apps/insights/src/observations.ts index e7a04e8e2..2140f00ff 100644 --- a/apps/insights/src/observations.ts +++ b/apps/insights/src/observations.ts @@ -53,6 +53,9 @@ export type LatestInsightObservation = Pick< export interface DueOpenInvestigation extends LatestInsightObservation { evidence: string[]; + // Synthetic shadow observations have no persisted identity. + id?: string; + insightId?: string | null; } export function nextRecheckAt( @@ -158,6 +161,8 @@ export async function loadDueOpenInvestigation(params: { }): Promise { const rows = await db .selectDistinctOn([insightObservations.signalKey], { + id: insightObservations.id, + insightId: insightObservations.insightId, evidence: insightObservations.evidence, outcome: insightObservations.outcome, recheckAt: insightObservations.recheckAt, diff --git a/apps/insights/src/persistence.ts b/apps/insights/src/persistence.ts index 741bfa48b..fbe21c1a8 100644 --- a/apps/insights/src/persistence.ts +++ b/apps/insights/src/persistence.ts @@ -1,7 +1,22 @@ import type { BusinessScope } from "@databuddy/ai/lib/business-context"; import { assertBusinessScopeCurrent } from "./business-context"; -import { and, db, desc, eq, isNotNull, lte, or, sql } from "@databuddy/db"; -import { analyticsInsights, insightObservations } from "@databuddy/db/schema"; +import { + and, + db, + desc, + eq, + isNotNull, + isNull, + lte, + or, + sql, +} from "@databuddy/db"; +import { + analyticsInsights, + insightObservations, + organization, + websites, +} from "@databuddy/db/schema"; import { invalidateAgentContextSnapshotsForWebsite, invalidateInsightsCachesForOrganization, @@ -10,9 +25,157 @@ import type { InvestigationOutcome, InvestigationSignal, } from "@databuddy/shared/insights"; +import { organizationBusinessContextSchema } from "@databuddy/shared/organization-business-context"; import { randomUUIDv7 } from "bun"; +import { z } from "zod"; import { normalizedErrorSubject } from "./investigation"; import { captureInsightsError, emitInsightsEvent } from "./lib/evlog-insights"; +import { measurementPlanKey } from "./measurement-plan"; +import type { DueOpenInvestigation } from "./observations"; + +export async function retireObsoleteRetentionObservation(params: { + asOf: Date; + domain: string; + observation: DueOpenInvestigation; + organizationId: string; + websiteId: string; +}): Promise { + const { observation } = params; + const signalKey = observation.signal.signalKey; + const insightId = observation.insightId; + if (!(signalKey.startsWith("retention:") && observation.id && insightId)) { + return false; + } + const retired = await db.transaction(async (tx) => { + // Match settings-save lock order and hold the canonical definition stable + // through the transition. A failed read must roll back, never imply removal. + const [owner] = await tx + .select({ metadata: organization.metadata }) + .from(organization) + .where(eq(organization.id, params.organizationId)) + .for("no key update"); + const [site] = await tx + .select({ id: websites.id }) + .from(websites) + .where( + and( + eq(websites.id, params.websiteId), + eq(websites.organizationId, params.organizationId), + eq(websites.domain, params.domain), + isNull(websites.deletedAt) + ) + ) + .for("update"); + if (!(owner?.metadata && site)) { + return false; + } + const { businessContext } = z + .object({ businessContext: organizationBusinessContextSchema.optional() }) + .parse(JSON.parse(owner.metadata)); + const profile = businessContext?.profile; + if ( + !profile?.measurementPlans || + Date.parse(profile.updatedAt) > params.asOf.getTime() || + profile.measurementPlans.some( + (plan) => + plan.websiteId === params.websiteId && + plan.domain === params.domain && + measurementPlanKey(plan) === signalKey + ) + ) { + return false; + } + const scope = and( + eq(analyticsInsights.id, insightId), + eq(analyticsInsights.organizationId, params.organizationId), + eq(analyticsInsights.websiteId, params.websiteId), + eq(analyticsInsights.subjectKey, signalKey), + eq(analyticsInsights.status, "open"), + lte(analyticsInsights.createdAt, params.asOf) + ); + const [current] = await tx + .select({ id: analyticsInsights.id }) + .from(analyticsInsights) + .where(scope) + .for("update"); + if (!current) { + return false; + } + const [latest] = await tx + .select() + .from(insightObservations) + .where( + and( + eq(insightObservations.organizationId, params.organizationId), + eq(insightObservations.websiteId, params.websiteId), + eq(insightObservations.signalKey, signalKey) + ) + ) + .orderBy( + desc(insightObservations.asOf), + desc(insightObservations.createdAt) + ) + .limit(1); + if ( + latest?.id !== observation.id || + latest.insightId !== current.id || + latest.asOf > params.asOf || + latest.createdAt > params.asOf || + latest.recheckAt > params.asOf || + latest.outcome.next.type === "resolve" + ) { + return false; + } + const reason = + "The saved activation and return definition was removed or changed. This investigation's measurement no longer applies; recovery was not measured."; + await tx + .update(analyticsInsights) + .set({ + // Supersede even an in-flight write with this exact snapshot time. + // The existing UPDATE/UPSERT fences both compare createdAt with <=. + createdAt: new Date(params.asOf.getTime() + 1), + status: "resolved", + resolvedAt: params.asOf, + resolvedReason: "stale", + }) + .where(scope); + await tx.insert(insightObservations).values({ + id: randomUUIDv7(), + insightId: current.id, + organizationId: params.organizationId, + websiteId: params.websiteId, + signalKey, + signal: latest.signal, + evidence: [reason], + outcome: { + title: latest.outcome.title, + summary: reason, + evidence: [reason], + rootCause: null, + impact: null, + publish: false, + next: { type: "resolve", reason }, + }, + asOf: params.asOf, + recheckAt: params.asOf, + }); + return true; + }); + if (retired) { + try { + await Promise.all([ + invalidateInsightsCachesForOrganization(params.organizationId), + invalidateAgentContextSnapshotsForWebsite(params.websiteId), + ]); + } catch (error) { + captureInsightsError(error, "generation.cache_invalidation.failed", { + organization_id: params.organizationId, + website_id: params.websiteId, + }); + } + } + return retired; +} export interface WebsiteInvestigation { id: string; diff --git a/apps/insights/src/retention-retirement.integration.test.ts b/apps/insights/src/retention-retirement.integration.test.ts new file mode 100644 index 000000000..23edad6d3 --- /dev/null +++ b/apps/insights/src/retention-retirement.integration.test.ts @@ -0,0 +1,661 @@ +import { randomUUID } from "node:crypto"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, +} from "bun:test"; +import type { executeQuery } from "@databuddy/ai/query"; +import { db, eq, inArray, shutdownPostgres } from "@databuddy/db"; +import { + analyticsInsights, + insightObservations, + insightRuns, + organization, + websites, +} from "@databuddy/db/schema"; +import { shutdownRedis } from "@databuddy/redis"; +import { saveOrganizationBusinessProfile } from "@databuddy/services/organization-business-context"; +import type { BusinessMeasurementPlan } from "@databuddy/shared/organization-business-context"; +import type { + InvestigationOutcome, + InvestigationSignal, +} from "@databuddy/shared/insights"; +import { parseInvestigationOutcome } from "@databuddy/shared/insights"; +import dayjs from "dayjs"; +import { + discoverWebsiteSignals, + type InvestigationSources, + remeasureStoredSignal, +} from "./generation"; +import { detectRetentionSignals, measurementPlanKey } from "./measurement-plan"; +import { prepareInvestigation } from "./investigation"; +import { + loadDueOpenInvestigation, + loadLatestSignalObservations, +} from "./observations"; +import { + persistInvestigation, + retireObsoleteRetentionObservation, +} from "./persistence"; + +// Run this file alone with env -i and --no-env-file. Only this synthetic DB is allowed. +const databaseUrl = + "postgresql://postgres:synthetic-only@localhost:16553/business_context_settings"; +describe("obsolete retention observations in synthetic PostgreSQL", () => { + let organizationId: string; + let other: string; + let websiteId: string; + let insightId: string; + let observationId: string; + let plan: BusinessMeasurementPlan; + let signal: InvestigationSignal; + let asOf: Date; + let revision: number; + const domain = "reports.example.com"; + const outcome: InvestigationOutcome = { + title: "Report return declined", + summary: "Fewer identified profiles returned after sharing a report.", + evidence: ["80/200 profiles returned, previously 160/200."], + rootCause: null, + impact: null, + publish: true, + next: { type: "ask", question: "Was the report flow changed?" }, + }; + beforeAll(() => { + if ( + process.env.DATABASE_URL !== databaseUrl || + process.env.REDIS_URL !== "redis://localhost:16554" || + process.env.BULLMQ_REDIS_URL !== "redis://localhost:16554" + ) { + throw new Error( + "Use only synthetic PostgreSQL at localhost:16553/business_context_settings and Redis at localhost:16554" + ); + } + }); + + const save = async (measurementPlans: BusinessMeasurementPlan[]) => { + const saved = await saveOrganizationBusinessProfile({ + organizationId, + revision, + content: "Synthetic report sharing service", + measurementPlans, + updatedBy: "synthetic-owner", + }); + revision = saved.profile?.revision ?? 0; + }; + const scope = () => ({ organizationId, websiteId, asOf }); + const rows = () => + db + .select() + .from(insightObservations) + .where(eq(insightObservations.websiteId, websiteId)); + const projection = async () => + ( + await db + .select() + .from(analyticsInsights) + .where(eq(analyticsInsights.id, insightId)) + )[0]; + const due = async () => { + const value = await loadDueOpenInvestigation(scope()); + if (!value) throw new Error("Expected a synthetic due observation"); + return value; + }; + const retire = async ( + overrides: Partial< + Parameters[0] + > = {} + ) => + retireObsoleteRetentionObservation({ + ...scope(), + domain, + observation: overrides.observation ?? (await due()), + ...overrides, + }); + + beforeEach(async () => { + organizationId = `retirement-${randomUUID()}`; + other = `retirement-${randomUUID()}`; + websiteId = `retirement-${randomUUID()}`; + insightId = randomUUID(); + observationId = randomUUID(); + asOf = new Date(Date.now() + 60_000); + revision = 0; + plan = { + websiteId, + domain, + name: "Report return", + activationEvent: "report_shared", + returnEvent: "report_opened", + horizonDays: 7, + }; + await db.insert(organization).values( + [organizationId, other].map((id) => ({ + id, + name: "Synthetic retirement", + slug: id, + createdAt: new Date(), + })) + ); + await db.insert(websites).values({ + id: websiteId, + organizationId, + domain, + name: "Synthetic reports", + }); + await save([plan]); + const previous = new Date(asOf.getTime() - 86_400_000); + signal = { + signalKey: measurementPlanKey(plan), + entity: { + type: "cohort", + id: measurementPlanKey(plan), + label: plan.name, + }, + metric: { + label: "Identified retention", + current: 40, + previous: 80, + format: "percent", + }, + changePercent: -50, + severity: "warning", + sentiment: "negative", + period: { + current: { from: "2026-08-25", to: "2026-08-31" }, + previous: { from: "2026-08-18", to: "2026-08-24" }, + }, + }; + await db.insert(analyticsInsights).values({ + id: insightId, + organizationId, + websiteId, + subjectKey: signal.signalKey, + title: outcome.title, + description: outcome.summary, + severity: "warning", + sentiment: "negative", + createdAt: previous, + }); + await db.insert(insightObservations).values({ + id: observationId, + organizationId, + websiteId, + insightId, + signalKey: signal.signalKey, + signal, + outcome, + evidence: outcome.evidence, + asOf: previous, + createdAt: previous, + recheckAt: previous, + }); + }); + afterEach(async () => { + // Delete only this test's organizations and their cascading synthetic fixtures. + await db + .delete(organization) + .where(inArray(organization.id, [organizationId, other])); + }); + afterAll(async () => { + await Promise.all([shutdownPostgres(), shutdownRedis()]); + }); + + const query = + (eligible = 200, incomplete = 0): typeof executeQuery => + async (request) => { + const today = dayjs(asOf).tz("UTC").startOf("day"); + const currentFrom = today.subtract(15, "day").format("YYYY-MM-DD"); + const retained = Math.floor( + eligible * (request.from === currentFrom ? 0.4 : 0.8) + ); + const row = { + cohort_from: request.from, + cohort_to: request.to, + observation_end: today.subtract(1, "day").format("YYYY-MM-DD"), + cohort_start: dayjs.tz(request.from, "UTC").toISOString(), + cohort_end: dayjs.tz(request.to, "UTC").add(1, "day").toISOString(), + observed_before: today.toISOString(), + timezone: "UTC", + horizon_days: 7, + identity_basis: "direct_profile_id", + activation_basis: "first_in_cohort_window", + activated_profiles: eligible + incomplete, + eligible_profiles: eligible, + retained_profiles: retained, + not_retained_profiles: eligible - retained, + incomplete_profiles: incomplete, + activation_events: eligible + incomplete, + identified_activation_events: eligible + incomplete, + unidentified_activation_events: 0, + }; + return [ + { ...row, row_type: "overall", cohort_date: null }, + { ...row, row_type: "cohort", cohort_date: request.from }, + ]; + }; + const discover = ( + retention: NonNullable< + Parameters[4] + >["retention"] = { query: query() }, + mode: "production" | "shadow" = "production", + overrides: Partial = {} + ) => { + const sources: InvestigationSources = { + loadDueInvestigation: loadDueOpenInvestigation, + loadObservations: loadLatestSignalObservations, + remeasureSignal: (params, prior, today, abortSignal) => + remeasureStoredSignal(params, prior, today, abortSignal, { retention }), + detectMetricSignals: async () => [], + detectDefinitionSignals: async () => [], + detectRouteHealthSignals: async () => [], + detectRetentionSignals: async () => [], + fetchAnnotations: async () => [], + loadHistory: async () => [], + loadOtherOpenWork: async () => [], + loadErrorCustomerImpact: async () => null, + loadRouteVitalContinuation: async () => null, + investigateSignal: async () => { + throw new Error("Discovery must not run a model"); + }, + ...overrides, + }; + return discoverWebsiteSignals( + { ...scope(), domain, timezone: "UTC" }, + { mode, sources } + ); + }; + + it.each([ + { activationEvent: "report_published" }, + { returnEvent: "report_reopened" }, + { activationEvent: "report_published", returnEvent: "report_reopened" }, + { namespace: "production" }, + { horizonDays: 30 as const }, + ])("retires only the old due case after saved selectors change: %j", async (changes) => { + await save([{ ...plan, ...changes }]); + const result = await discover({ + query: async () => { + throw new Error("Obsolete selectors must not be measured"); + }, + }); + expect(result).toMatchObject({ + kind: "empty", + artifact: { status: "no_signals" }, + }); + expect(await projection()).toMatchObject({ + status: "resolved", + resolvedReason: "stale", + }); + const history = await rows(); + expect(history).toHaveLength(2); + expect(history.find((row) => row.id === observationId)?.outcome).toEqual( + outcome + ); + expect(history.find((row) => row.id !== observationId)).toMatchObject({ + signal, + insightId, + outcome: { publish: false, next: { type: "resolve" }, rootCause: null }, + }); + expect( + history.find((row) => row.id !== observationId)?.outcome.summary + ).toContain("recovery was not measured"); + expect( + parseInvestigationOutcome( + history.find((row) => row.id !== observationId)?.outcome + ) + ).not.toBeNull(); + expect(await loadDueOpenInvestigation(scope())).toBeNull(); + await discover(); + expect(await rows()).toHaveLength(2); + }); + it("preserves the replacement definition's open case on the same website", async () => { + const replacement = { ...plan, returnEvent: "report_reopened" }; + await save([replacement]); + const replacementId = randomUUID(); + const replacementSignal = { + ...signal, + signalKey: measurementPlanKey(replacement), + }; + const recent = new Date(asOf.getTime() - 60_000); + await db.insert(analyticsInsights).values({ + id: replacementId, + organizationId, + websiteId, + subjectKey: replacementSignal.signalKey, + title: outcome.title, + description: outcome.summary, + severity: "warning", + sentiment: "negative", + createdAt: recent, + }); + await db.insert(insightObservations).values({ + id: randomUUID(), + organizationId, + websiteId, + insightId: replacementId, + signalKey: replacementSignal.signalKey, + signal: replacementSignal, + outcome, + asOf: recent, + createdAt: recent, + recheckAt: new Date(asOf.getTime() + 86_400_000), + }); + await discover(); + expect(await projection()).toMatchObject({ + status: "resolved", + resolvedReason: "stale", + }); + const [replacementCase] = await db + .select() + .from(analyticsInsights) + .where(eq(analyticsInsights.id, replacementId)); + expect(replacementCase).toMatchObject({ + status: "open", + resolvedReason: null, + }); + expect(await rows()).toHaveLength(3); + }); + it("excludes a retired key even if a parallel detector read its old definition", async () => { + const detected = await detectRetentionSignals( + { websiteId, lookbackDays: 7, timezone: "UTC" }, + dayjs(asOf), + undefined, + { query: query() } + ); + expect(detected).toHaveLength(1); + await save([]); + expect( + await discover(undefined, "production", { + detectRetentionSignals: async () => detected, + }) + ).toMatchObject({ + kind: "empty", + artifact: { status: "no_signals" }, + }); + expect(await projection()).toMatchObject({ + status: "resolved", + resolvedReason: "stale", + }); + expect(await rows()).toHaveLength(2); + }); + it("retires a removed definition and leaves no endlessly deferred case", async () => { + await save([]); + expect(await discover()).toMatchObject({ + kind: "empty", + artifact: { status: "no_signals" }, + }); + expect(await projection()).toMatchObject({ + status: "resolved", + resolvedReason: "stale", + }); + expect(await loadDueOpenInvestigation(scope())).toBeNull(); + }); + it("remeasures a label-only rename without retiring or rewriting its history", async () => { + await save([{ ...plan, name: "Renamed report return" }]); + expect(await discover()).toMatchObject({ kind: "signals" }); + expect(await retire()).toBe(false); + expect(await projection()).toMatchObject({ + status: "open", + resolvedReason: null, + }); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it.each([ + { eligible: 49, incomplete: 0 }, + { eligible: 200, incomplete: 1 }, + ])("keeps unavailable cohorts open: %j", async ({ eligible, incomplete }) => { + expect( + await discover({ query: query(eligible, incomplete) }) + ).toMatchObject({ kind: "empty", artifact: { status: "deferred" } }); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it.each([ + "settings", + "analytics", + ])("leaves the persisted case unchanged after a transient %s failure", async (source) => { + const failure = async () => { + throw new Error("Synthetic read unavailable"); + }; + await expect( + discover( + source === "settings" ? { readPlan: failure } : { query: failure } + ) + ).rejects.toThrow("Synthetic read unavailable"); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it("does not mistake an empty analytics response for a removed definition", async () => { + await expect(discover({ query: async () => [] })).rejects.toThrow(); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it.each([ + null, + { + content: "Synthetic context", + origin: "team", + revision: 3, + updatedAt: new Date().toISOString(), + updatedBy: "synthetic", + sources: [], + sourceWebsiteId: null, + }, + ])("defers when the canonical profile or selector list is unavailable", async (profile) => { + await db + .update(organization) + .set({ + metadata: JSON.stringify({ + businessContext: { profile, generation: null }, + }), + }) + .where(eq(organization.id, organizationId)); + expect(await discover()).toMatchObject({ + kind: "empty", + artifact: { status: "deferred" }, + }); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it("does not apply a newer canonical definition to a historical recheck", async () => { + await save([]); + asOf = new Date(Date.now() - 60_000); + expect(await discover()).toMatchObject({ + kind: "empty", + artifact: { status: "deferred" }, + }); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it("keeps shadow discovery read-only for an obsolete definition", async () => { + await save([]); + expect(await discover(undefined, "shadow")).toMatchObject({ + kind: "empty", + artifact: { status: "deferred" }, + }); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it("refuses a stale observation pointer or foreign tenant, website, or domain", async () => { + await save([]); + const observation = await due(); + for (const overrides of [ + { observation: { ...observation, id: randomUUID() } }, + { observation: { ...observation, insightId: randomUUID() } }, + { organizationId: other }, + { websiteId: randomUUID() }, + { domain: "other.example.com" }, + ]) + expect(await retire(overrides)).toBe(false); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it("rechecks the definition at persistence if it is restored during detection", async () => { + await save([]); + expect( + await discover({ + readPlan: async () => { + await save([plan]); + return null; + }, + }) + ).toMatchObject({ kind: "empty", artifact: { status: "deferred" } }); + expect((await due()).id).toBe(observationId); + expect(await rows()).toHaveLength(1); + }); + it.each([ + { offset: -1, next: "ask", deduped: false }, + { offset: 0, next: "ask", deduped: false }, + { offset: 0, next: "act", deduped: false }, + { offset: 0, next: "ask", deduped: true }, + { offset: 0, next: "act", deduped: true }, + ])("blocks older and equal-snapshot in-flight writes: %j", async ({ + offset, + next, + deduped, + }) => { + const runId = randomUUID(); + await db.insert(insightRuns).values({ + id: runId, + organizationId, + reason: "scheduled", + status: "running", + }); + if (deduped) { + await db + .update(analyticsInsights) + .set({ dedupeKey: `${websiteId}|${signal.signalKey}` }) + .where(eq(analyticsInsights.id, insightId)); + } + await save([]); + expect(await retire()).toBe(true); + await expect( + persistInvestigation({ + investigation: { + id: insightId, + signal, + outcome: { + ...outcome, + next: + next === "ask" + ? outcome.next + : { + type: "act", + action: "Review the synthetic report flow", + target: "Synthetic reports", + verification: "Synthetic return recovers", + }, + }, + websiteId, + websiteDomain: domain, + websiteName: "Synthetic reports", + }, + organizationId, + notNewerThan: new Date(asOf.getTime() + offset), + recheckAt: asOf, + runId, + timezone: "UTC", + }) + ).rejects.toThrow("changed while scheduled analysis was running"); + expect(await projection()).toMatchObject({ + status: "resolved", + resolvedReason: "stale", + }); + expect(await rows()).toHaveLength(2); + }); + it.each([ + false, + true, + ])("allows later measured work to reopen a restored definition (deduped: %s)", async (deduped) => { + if (deduped) { + await db + .update(analyticsInsights) + .set({ dedupeKey: `${websiteId}|${signal.signalKey}` }) + .where(eq(analyticsInsights.id, insightId)); + } + await save([]); + expect(await retire()).toBe(true); + await save([plan]); + const after = new Date(asOf.getTime() + 2); + const detected = await detectRetentionSignals( + { websiteId, lookbackDays: 7, timezone: "UTC" }, + dayjs(after), + undefined, + { query: query() } + ); + expect(detected).toHaveLength(1); + const measured = prepareInvestigation(detected[0], 7); + const runId = randomUUID(); + await db + .insert(insightRuns) + .values({ + id: runId, + organizationId, + reason: "scheduled", + status: "running", + }); + expect( + await persistInvestigation({ + investigation: { + id: insightId, + signal: measured.signal, + outcome, + websiteId, + websiteDomain: domain, + websiteName: "Synthetic reports", + }, + evidence: measured.evidence, + organizationId, + notNewerThan: after, + recheckAt: after, + runId, + timezone: "UTC", + }) + ).toMatchObject({ id: insightId }); + expect(await projection()).toMatchObject({ + status: "open", + resolvedReason: null, + }); + expect(await rows()).toHaveLength(3); + }); + it("does not close a case with a newer observation, including one beyond this scan", async () => { + await save([]); + const observation = await due(); + const newer = new Date(asOf.getTime() + 1); + await db.insert(insightObservations).values({ + id: randomUUID(), + organizationId, + websiteId, + insightId, + signalKey: signal.signalKey, + signal, + outcome, + asOf: newer, + createdAt: newer, + recheckAt: newer, + }); + expect(await retire({ observation })).toBe(false); + expect(await projection()).toMatchObject({ + status: "open", + resolvedReason: null, + }); + expect(await rows()).toHaveLength(2); + }); + it("appends exactly one transition when two workers retire the same observation", async () => { + await save([]); + const observation = await due(); + const results = await Promise.all([ + retire({ observation }), + retire({ observation }), + ]); + expect(results.sort()).toEqual([false, true]); + expect(await rows()).toHaveLength(2); + expect(await loadDueOpenInvestigation(scope())).toBeNull(); + }); +}); diff --git a/package.json b/package.json index e4cf562f2..a1771aa7a 100644 --- a/package.json +++ b/package.json @@ -36,8 +36,8 @@ "dev": "dotenv -- turbo run dev --env-mode=loose", "start": "NODE_ENV=production dotenv -- turbo run start --env-mode=loose", "test": "dotenv -- turbo run test", - "test:watch": "dotenv -- bun test --watch ./apps", - "test:coverage": "dotenv -- bun test --coverage ./apps", + "test:watch": "dotenv -- bun test --watch ./apps --path-ignore-patterns='**/*.integration.test.ts'", + "test:coverage": "dotenv -- bun test --coverage ./apps --path-ignore-patterns='**/*.integration.test.ts'", "lint": "bunx ultracite check && bun run lint:policies", "knip": "knip", "lint:policies": "bunx tsc --project scripts/tsconfig.json && bun test scripts/lint-policy.test.ts && bun scripts/lint-policy.ts", diff --git a/packages/services/src/measurement-plan.integration.test.ts b/packages/services/src/measurement-plan.integration.test.ts index 84484942d..94d8b6cd0 100644 --- a/packages/services/src/measurement-plan.integration.test.ts +++ b/packages/services/src/measurement-plan.integration.test.ts @@ -209,6 +209,52 @@ integration("measurement plan storage in synthetic PostgreSQL", () => { expect(read.history).toEqual([original.profile]); }); + test.each([ + "transferred", + "deleted", + "missing", + "changed domain", + ] as const)("validates inherited plans after their website is %s", async (binding) => { + await save({ revision: 0, content: "Original", measurementPlans: plans }); + const generationId = await ready(); + if (binding === "transferred") { + await db.delete(websites).where(eq(websites.id, foreignId)); + await db + .update(websites) + .set({ organizationId: other }) + .where(eq(websites.id, secondaryId)); + } else if (binding === "deleted") { + await db + .update(websites) + .set({ deletedAt: new Date() }) + .where(eq(websites.id, secondaryId)); + } else if (binding === "missing") { + await db.delete(websites).where(eq(websites.id, secondaryId)); + } else { + await db + .update(websites) + .set({ domain: "changed.example.com" }) + .where(eq(websites.id, secondaryId)); + } + const before = await metadata(); + const foreignBefore = await metadata(other); + for (const candidate of [ + { revision: 1, content: "Text-only edit", teamContext }, + { revision: 1, content: draft.content, generationId }, + ]) { + await expect(save(candidate)).rejects.toMatchObject({ code: "CONFLICT" }); + expect(await metadata()).toBe(before); + expect(await metadata(other)).toBe(foreignBefore); + } + const repaired = await save({ + revision: 1, + content: "Remove the stale definition", + measurementPlans: [plans[0]], + }); + expect(repaired.profile.measurementPlans).toEqual([plans[0]]); + expect(repaired.profile.revision).toBe(2); + }); + test("an explicit empty array clears plans and a later omission keeps them cleared", async () => { const original = await save({ revision: 0, diff --git a/packages/services/src/organization-business-context.ts b/packages/services/src/organization-business-context.ts index c0b45193a..eb50afffe 100644 --- a/packages/services/src/organization-business-context.ts +++ b/packages/services/src/organization-business-context.ts @@ -300,7 +300,7 @@ export async function saveOrganizationBusinessProfile(input: { await validateMeasurementBindings( tx, input.organizationId, - input.measurementPlans + measurementPlans ); const unchangedDraft = generated?.draft?.content === content; const unchangedSaved = !generated && current.profile?.content === content;