feat(alerts): usage-limit and data-health notifications

apps/api/src/controllers/webhook.controller.ts

@@ -317,6 +317,12 @@ async function syncSubscriptionToOrg(
       organization.subscriptionPeriodEventsLimit < subscriptionPeriodEventsLimit
         ? null
         : undefined,
+    // A raised limit re-arms the usage alerts for the new headroom.
+    ...(typeof subscriptionPeriodEventsLimit === 'number' &&
+    typeof organization.subscriptionPeriodEventsLimit === 'number' &&
+    organization.subscriptionPeriodEventsLimit < subscriptionPeriodEventsLimit
+      ? { usageWarningSentAt: null, usageExceededSentAt: null }
+      : {}),
   };
 
   const changes = diffOrganizationFields(
@@ -444,6 +450,9 @@ export async function polarWebhook(
           data: {
             subscriptionPeriodEventsCount: 0,
             subscriptionPeriodEventsCountExceededAt: null,
+            // New cycle — the usage alerts may fire again.
+            usageWarningSentAt: null,
+            usageExceededSentAt: null,
           },
         });
 

apps/start/src/routes/_app.$organizationId.tsx

@@ -162,12 +162,31 @@ function Component() {
           </LinkButton>
         </Alert>
       )}
+      {organization.isActive &&
+        !organization.isExceeded &&
+        organization.subscriptionPeriodEventsLimit > 0 &&
+        organization.subscriptionPeriodEventsCount >=
+          organization.subscriptionPeriodEventsLimit * 0.8 && (
+          <Alert
+            title="Approaching your events limit"
+            description={`You've used ${Math.round((organization.subscriptionPeriodEventsCount / organization.subscriptionPeriodEventsLimit) * 100)}% of your ${organization.subscriptionPeriodEventsLimit.toLocaleString()} monthly events. If you go over, we keep collecting your events but charts pause until you upgrade.`}
+          >
+            <LinkButton
+              to="/$organizationId/billing"
+              params={{
+                organizationId: organizationId,
+              }}
+            >
+              See plans
+            </LinkButton>
+          </Alert>
+        )}
       {organization.subscriptionPeriodEventsCountExceededAt &&
         organization.isActive &&
         organization.isExceeded && (
           <Alert
             title="Events limit exceeded"
-            description={`Your subscription has exceeded the limit on ${format(organization.subscriptionPeriodEventsCountExceededAt, 'PPP')}`}
+            description={`You hit your monthly events limit on ${format(organization.subscriptionPeriodEventsCountExceededAt, 'PPP')}. We're still collecting your events — nothing is lost — but charts won't show new data until you upgrade or your next cycle starts.`}
           >
             <LinkButton
               to="/$organizationId/billing"

apps/worker/src/boot-cron.ts

@@ -113,6 +113,11 @@ export async function bootCron() {
       type: 'weeklyDigest',
       pattern: '0 8 * * 1', // Mondays 08:00 UTC — weekly analytics digest email
     },
+    {
+      name: 'dataHealth',
+      type: 'dataHealth',
+      pattern: '30 7 * * *', // Daily 07:30 UTC — no-data / data-stopped rescue emails
+    },
   ];
 
   if (process.env.SELF_HOSTED && process.env.NODE_ENV === 'production') {

apps/worker/src/boot-debug.ts

@@ -30,6 +30,7 @@ const CRON_TYPES = [
   'sessionVacuum',
   'insightCleanup',
   'weeklyDigest',
+  'dataHealth',
 ] as const satisfies readonly CronQueueType[];
 
 function escapeHtml(value: string) {

apps/worker/src/jobs/cron.data-health.ts

@@ -0,0 +1,187 @@
+import { db, getLastEventPerProject } from '@openpanel/db';
+import { sendEmail } from '@openpanel/email';
+import { logger as baseLogger } from '@/utils/logger';
+
+const logger = baseLogger.child({ job: 'data-health' });
+
+const DAY_MS = 24 * 60 * 60 * 1000;
+// A brand-new project gets 48h to send its first event before we reach out.
+const NO_DATA_AFTER_MS = 2 * DAY_MS;
+// An active project whose newest event is older than this counts as stalled.
+const STALLED_AFTER_MS = 7 * DAY_MS;
+
+interface OrgAlert {
+  organizationId: string;
+  noData: { id: string; name: string }[];
+  stalled: { id: string; name: string; lastEventAt: Date }[];
+}
+
+async function recipientsForOrg(organizationId: string) {
+  const members = await db.member.findMany({
+    where: {
+      organizationId,
+      user: { deletedAt: null },
+    },
+    include: { user: { select: { email: true, firstName: true } } },
+  });
+  const seen = new Map<string, string | undefined>();
+  for (const member of members) {
+    if (member.user?.email && !seen.has(member.user.email)) {
+      seen.set(member.user.email, member.user.firstName ?? undefined);
+    }
+  }
+  return seen;
+}
+
+/**
+ * Daily rescue emails for paying/trialing orgs whose tracking is broken:
+ * projects that never received an event (48h grace) and projects whose event
+ * flow stalled for 7+ days. A silently broken install reads as "the product
+ * stopped working" — a working install is the cheapest retention there is.
+ *
+ * Dedupe: `noDataNotifiedAt` is sent once per project; `dataStoppedNotifiedAt`
+ * is compared against the newest event, so data resuming and stalling again
+ * notifies again without any clearing step.
+ */
+export async function dataHealthCronJob() {
+  if (process.env.SELF_HOSTED === 'true') {
+    return;
+  }
+
+  const now = Date.now();
+  const lastEventByProject = await getLastEventPerProject();
+
+  // Prefilter on the raw status column (computed fields can't be used in
+  // `where`), then refine with the canonical subscription state below.
+  const projects = await db.project.findMany({
+    where: {
+      deleteAt: null,
+      organization: { subscriptionStatus: { in: ['active', 'trialing'] } },
+    },
+    select: {
+      id: true,
+      name: true,
+      createdAt: true,
+      organizationId: true,
+      noDataNotifiedAt: true,
+      dataStoppedNotifiedAt: true,
+      organization: { select: { id: true, subscriptionState: true } },
+    },
+  });
+
+  const byOrg = new Map<string, OrgAlert>();
+
+  for (const project of projects) {
+    const state = project.organization.subscriptionState;
+    if (state !== 'active' && state !== 'trialing') {
+      continue;
+    }
+
+    const lastEventAt = lastEventByProject.get(project.id);
+
+    if (!lastEventAt) {
+      // Never received an event. One notice per project, after the grace
+      // period. (The onboarding drip already nudges brand-new orgs, so this
+      // mainly catches additional projects and broken installs.)
+      const oldEnough = now - project.createdAt.getTime() > NO_DATA_AFTER_MS;
+      if (oldEnough && !project.noDataNotifiedAt) {
+        const entry = byOrg.get(project.organizationId) ?? {
+          organizationId: project.organizationId,
+          noData: [],
+          stalled: [],
+        };
+        entry.noData.push({ id: project.id, name: project.name });
+        byOrg.set(project.organizationId, entry);
+      }
+      continue;
+    }
+
+    const stalled = now - lastEventAt.getTime() > STALLED_AFTER_MS;
+    const alreadyNotifiedForThisStall =
+      project.dataStoppedNotifiedAt &&
+      project.dataStoppedNotifiedAt > lastEventAt;
+    if (stalled && !alreadyNotifiedForThisStall) {
+      const entry = byOrg.get(project.organizationId) ?? {
+        organizationId: project.organizationId,
+        noData: [],
+        stalled: [],
+      };
+      entry.stalled.push({
+        id: project.id,
+        name: project.name,
+        lastEventAt,
+      });
+      byOrg.set(project.organizationId, entry);
+    }
+  }
+
+  let emailsSent = 0;
+
+  for (const alert of byOrg.values()) {
+    try {
+      const recipients = await recipientsForOrg(alert.organizationId);
+      if (recipients.size === 0) {
+        continue;
+      }
+
+      const dashboardUrl = `${process.env.DASHBOARD_URL ?? 'https://dashboard.openpanel.dev'}/${alert.organizationId}`;
+
+      if (alert.noData.length > 0) {
+        for (const [to, firstName] of recipients) {
+          await sendEmail('tracking-no-data', {
+            to,
+            data: {
+              firstName,
+              projectNames: alert.noData.map((p) => p.name),
+              dashboardUrl,
+            },
+          });
+          emailsSent++;
+        }
+        await db.project.updateMany({
+          where: { id: { in: alert.noData.map((p) => p.id) } },
+          data: { noDataNotifiedAt: new Date() },
+        });
+      }
+
+      if (alert.stalled.length > 0) {
+        const newestLastEvent = alert.stalled
+          .map((p) => p.lastEventAt)
+          .sort((a, b) => b.getTime() - a.getTime())[0];
+        for (const [to, firstName] of recipients) {
+          await sendEmail('tracking-data-stopped', {
+            to,
+            data: {
+              firstName,
+              projectNames: alert.stalled.map((p) => p.name),
+              lastEventDate: newestLastEvent?.toLocaleDateString('en-US', {
+                month: 'long',
+                day: 'numeric',
+              }),
+              dashboardUrl,
+            },
+          });
+          emailsSent++;
+        }
+        await db.project.updateMany({
+          where: { id: { in: alert.stalled.map((p) => p.id) } },
+          data: { dataStoppedNotifiedAt: new Date() },
+        });
+      }
+    } catch (err) {
+      logger.error(
+        { err, organizationId: alert.organizationId },
+        'Data-health alert failed'
+      );
+    }
+  }
+
+  logger.info(
+    {
+      projects: projects.length,
+      organizations: byOrg.size,
+      emailsSent,
+    },
+    'Data-health check complete'
+  );
+}

apps/worker/src/jobs/cron.ts

@@ -9,6 +9,7 @@ import {
 import type { CronQueuePayload } from '@openpanel/queue';
 import type { Job } from 'bullmq';
 import { cohortRefreshCronJob } from './cron.cohort-refresh';
+import { dataHealthCronJob } from './cron.data-health';
 import { jobDelete } from './cron.delete';
 import { insightCleanupCronJob } from './cron.insight-cleanup';
 import { weeklyDigestCronJob } from './cron.weekly-digest';
@@ -75,5 +76,8 @@ export async function cronJob(job: Job<CronQueuePayload>) {
     case 'weeklyDigest': {
       return await weeklyDigestCronJob();
     }
+    case 'dataHealth': {
+      return await dataHealthCronJob();
+    }
   }
 }

apps/worker/src/jobs/cron.weekly-digest.ts

@@ -6,7 +6,11 @@ import { logger as baseLogger } from '@/utils/logger';
 const logger = baseLogger.child({ job: 'weekly-digest' });
 
 const DAY_MS = 24 * 60 * 60 * 1000;
-const MIN_EVENTS = 5000;
+// Keep this low: the digest is our main "value without logging in" touchpoint,
+// and the old 5000 gate excluded customers on smaller plans — exactly the ones
+// who benefit most from the reminder. Zero-visitor weeks are still skipped per
+// send, so quiet projects don't get empty emails.
+const MIN_EVENTS = 100;
 const MAX_INSIGHTS = 5;
 
 type DigestData = EmailData<'weekly-digest'>;

apps/worker/src/jobs/notification.ts

@@ -1,6 +1,7 @@
 import type { Job } from 'bullmq';
 
 import { Prisma, db } from '@openpanel/db';
+import { sendEmail } from '@openpanel/email';
 import { sendDiscordNotification } from '@openpanel/integrations/src/discord';
 import { sendSlackNotification } from '@openpanel/integrations/src/slack';
 import { execute as executeJavaScriptTemplate } from '@openpanel/js-runtime';
@@ -29,6 +30,35 @@ export async function notificationJob(job: Job<NotificationQueuePayload>) {
       }
 
       if (notification.sendToEmail) {
+        const project = await db.project.findUniqueOrThrow({
+          where: { id: notification.projectId },
+          select: { name: true, organizationId: true },
+        });
+        const members = await db.member.findMany({
+          where: {
+            organizationId: project.organizationId,
+            user: { deletedAt: null },
+          },
+          include: { user: { select: { email: true } } },
+        });
+        const emails = new Set(
+          members.flatMap((member) =>
+            member.user?.email ? [member.user.email] : [],
+          ),
+        );
+        for (const to of emails) {
+          // Per-recipient unsubscribe (product_alerts category) is handled
+          // inside sendEmail.
+          await sendEmail('notification-rule', {
+            to,
+            data: {
+              title: notification.title,
+              message: notification.message,
+              projectName: project.name,
+              dashboardUrl: `${process.env.DASHBOARD_URL ?? 'https://dashboard.openpanel.dev'}/${project.organizationId}/${notification.projectId}`,
+            },
+          });
+        }
         return;
       }