From eee0d9b64a09323613387e45c1183c94fd84e3e0 Mon Sep 17 00:00:00 2001 From: iza <59828082+izadoesdev@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:33:39 +0300 Subject: [PATCH 1/6] fix(rpc): validate inherited measurement bindings --- .../src/measurement-plan.integration.test.ts | 46 +++++++++++++++++++ .../src/organization-business-context.ts | 2 +- 2 files changed, 47 insertions(+), 1 deletion(-) 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; From 8b51461de879fcff3a33e32c4d2442f824b38d6d Mon Sep 17 00:00:00 2001 From: iza <59828082+izadoesdev@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:33:40 +0300 Subject: [PATCH 2/6] style(dashboard): apply measurement typography utilities --- .../components/measurement-plan-editor.tsx | 32 ++++++++++--------- 1 file changed, 17 insertions(+), 15 deletions(-) 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.

)} From 25c00c1bdc73e64619e256240109f0d449f01bb1 Mon Sep 17 00:00:00 2001 From: iza <59828082+izadoesdev@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:35:05 +0300 Subject: [PATCH 3/6] docs(ci): require resolved PR feedback before merging --- .agents/skills/databuddy-internal/SKILL.md | 1 + AGENTS.md | 1 + 2 files changed, 2 insertions(+) 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. From 2d0754f022423a7b523cf61cbf65cd60ba06a375 Mon Sep 17 00:00:00 2001 From: iza <59828082+izadoesdev@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:48:31 +0300 Subject: [PATCH 4/6] fix(insights): retire obsolete retention observations --- apps/insights/src/generation.ts | 25 +- apps/insights/src/observations.ts | 5 + apps/insights/src/persistence.ts | 167 ++++- .../retention-retirement.integration.test.ts | 673 ++++++++++++++++++ 4 files changed, 864 insertions(+), 6 deletions(-) create mode 100644 apps/insights/src/retention-retirement.integration.test.ts 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..6ff0f2f31 --- /dev/null +++ b/apps/insights/src/retention-retirement.integration.test.ts @@ -0,0 +1,673 @@ +import { randomUUID } from "node:crypto"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + spyOn, +} 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 * as cache 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"; +const integration = + process.env.INSIGHTS_INTEGRATION_TESTS === "true" ? describe : describe.skip; + +integration("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?" }, + }; + const invalidations: ReturnType[] = []; + + beforeAll(() => { + if (process.env.DATABASE_URL !== databaseUrl) { + throw new Error( + "Use only the synthetic localhost:16553/business_context_settings database" + ); + } + invalidations.push( + spyOn(cache, "invalidateInsightsCachesForOrganization").mockResolvedValue( + { attempted: 2, failed: 0 } + ), + spyOn( + cache, + "invalidateAgentContextSnapshotsForWebsite" + ).mockResolvedValue(0) + ); + }); + + 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 () => { + for (const invalidation of invalidations) invalidation.mockRestore(); + await shutdownPostgres(); + }); + + 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(); + }); +}); From 16a72e77e5de1b9a4860fba1979345e1ae7559a1 Mon Sep 17 00:00:00 2001 From: iza <59828082+izadoesdev@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:59:51 +0300 Subject: [PATCH 5/6] test(insights): exercise retirement with real cache invalidation --- .../retention-retirement.integration.test.ts | 27 +++++-------------- 1 file changed, 7 insertions(+), 20 deletions(-) diff --git a/apps/insights/src/retention-retirement.integration.test.ts b/apps/insights/src/retention-retirement.integration.test.ts index 6ff0f2f31..a05f75bf9 100644 --- a/apps/insights/src/retention-retirement.integration.test.ts +++ b/apps/insights/src/retention-retirement.integration.test.ts @@ -7,7 +7,6 @@ import { describe, expect, it, - spyOn, } from "bun:test"; import type { executeQuery } from "@databuddy/ai/query"; import { db, eq, inArray, shutdownPostgres } from "@databuddy/db"; @@ -18,7 +17,6 @@ import { organization, websites, } from "@databuddy/db/schema"; -import * as cache from "@databuddy/redis"; import { saveOrganizationBusinessProfile } from "@databuddy/services/organization-business-context"; import type { BusinessMeasurementPlan } from "@databuddy/shared/organization-business-context"; import type { @@ -46,10 +44,7 @@ import { // 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"; -const integration = - process.env.INSIGHTS_INTEGRATION_TESTS === "true" ? describe : describe.skip; - -integration("obsolete retention observations in synthetic PostgreSQL", () => { +describe("obsolete retention observations in synthetic PostgreSQL", () => { let organizationId: string; let other: string; let websiteId: string; @@ -69,23 +64,16 @@ integration("obsolete retention observations in synthetic PostgreSQL", () => { publish: true, next: { type: "ask", question: "Was the report flow changed?" }, }; - const invalidations: ReturnType[] = []; - beforeAll(() => { - if (process.env.DATABASE_URL !== databaseUrl) { + 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 the synthetic localhost:16553/business_context_settings database" + "Use only synthetic PostgreSQL at localhost:16553/business_context_settings and Redis at localhost:16554" ); } - invalidations.push( - spyOn(cache, "invalidateInsightsCachesForOrganization").mockResolvedValue( - { attempted: 2, failed: 0 } - ), - spyOn( - cache, - "invalidateAgentContextSnapshotsForWebsite" - ).mockResolvedValue(0) - ); }); const save = async (measurementPlans: BusinessMeasurementPlan[]) => { @@ -213,7 +201,6 @@ integration("obsolete retention observations in synthetic PostgreSQL", () => { .where(inArray(organization.id, [organizationId, other])); }); afterAll(async () => { - for (const invalidation of invalidations) invalidation.mockRestore(); await shutdownPostgres(); }); From 66d1ed7e536547da5e1586f0c2afe870c21a4f8b Mon Sep 17 00:00:00 2001 From: iza <59828082+izadoesdev@users.noreply.github.com> Date: Wed, 9 Sep 2026 15:11:12 +0300 Subject: [PATCH 6/6] test(insights): close test resources and isolate integration selection --- apps/insights/src/retention-retirement.integration.test.ts | 3 ++- package.json | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/apps/insights/src/retention-retirement.integration.test.ts b/apps/insights/src/retention-retirement.integration.test.ts index a05f75bf9..23edad6d3 100644 --- a/apps/insights/src/retention-retirement.integration.test.ts +++ b/apps/insights/src/retention-retirement.integration.test.ts @@ -17,6 +17,7 @@ import { 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 { @@ -201,7 +202,7 @@ describe("obsolete retention observations in synthetic PostgreSQL", () => { .where(inArray(organization.id, [organizationId, other])); }); afterAll(async () => { - await shutdownPostgres(); + await Promise.all([shutdownPostgres(), shutdownRedis()]); }); const query = 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",