Weekly digest

// src/journeys/weekly-digest.ts
import { days, defineJourney, sendEmail } from "@hogsend/engine";
import { Events, Templates } from "./constants/index.js";

export const weeklyDigest = defineJourney({
  meta: {
    id: "weekly-digest",
    name: "Retention — Weekly activity digest",
    enabled: true,
    // Any report activity opens a window; the rest of the week folds in.
    trigger: { event: Events.REPORT_CREATED },
    entryLimit: "unlimited", // rolling: a fresh window opens after each flush
    // ctx.digest already collapses the week into one send, so the per-journey
    // min-gap must be off — a suppress >= the window would gap out each new
    // window's email against the previous one's.
    suppress: days(0),
    exitOn: [{ event: Events.USER_DELETED }],
  },

  run: async (user, ctx) => {
    // The first report enrolls; every report.created in the next 7 days is
    // absorbed by the active-enrollment guard and returned here at flush.
    // One execution, one email — not one per report.
    const digest = await ctx.digest({ window: days(7), label: "weekly" });

    // A 7-day window is a long wait, and unsubscribe doesn't exit the journey.
    if (!(await ctx.guard.isSubscribed())) return;

    // The "batch" recipe: ctx.digest only collects the window; grouping is
    // plain TypeScript over digest.events.
    const byProject = Object.groupBy(
      digest.events,
      (e) => String(e.properties?.projectId ?? "unknown"),
    );
    const projects = Object.entries(byProject).map(([projectId, events]) => ({
      projectId,
      count: events?.length ?? 0,
    }));

    await sendEmail({
      to: user.email,
      userId: user.id,
      journeyStateId: user.stateId,
      template: Templates.RETENTION_WEEKLY_DIGEST, // "retention-weekly-digest"
      subject: "Your week in review",
      journeyName: user.journeyName,
      props: { totalReports: digest.count, projects },
    });
  },
});

// src/journeys/constants/index.ts
export const Events = {
  REPORT_CREATED: "report.created",
  USER_DELETED: "user.deleted",
} as const;

export const Templates = {
  RETENTION_WEEKLY_DIGEST: "retention-weekly-digest",
} as const;

// src/workflows/weekly-digest.ts
import { contacts, userEvents } from "@hogsend/db";
import { hatchet } from "@hogsend/engine";
import { and, eq, gte, inArray, sql } from "drizzle-orm";
import { getContainer } from "../container.js";
import { Events, Templates } from "../journeys/constants/index.js";

const WINDOW_MS = 7 * 24 * 60 * 60 * 1000;

export const weeklyDigestTask = hatchet.task({
  name: "weekly-digest",
  // Mondays at 09:00 UTC — cron expressions evaluate in UTC, not per-user time.
  onCrons: ["0 9 * * 1"],
  retries: 1,
  executionTimeout: "30m",
  fn: async () => {
    const { db, emailService, logger } = getContainer();
    const since = new Date(Date.now() - WINDOW_MS);
    // One key per (user, weekly run): retries and re-runs can't double-send.
    const weekKey = new Date().toISOString().slice(0, 10);

    // One aggregate query: per-user counts over the window. Users with no
    // qualifying events never appear — the empty digest is structurally
    // impossible, not a filter you have to remember.
    const activity = await db
      .select({
        userId: userEvents.userId,
        reportsCreated: sql<number>`count(*) filter (where ${userEvents.event} = ${Events.REPORT_CREATED})`,
        reportsShared: sql<number>`count(*) filter (where ${userEvents.event} = ${Events.REPORT_SHARED})`,
      })
      .from(userEvents)
      .where(
        and(
          gte(userEvents.occurredAt, since),
          inArray(userEvents.event, [
            Events.REPORT_CREATED,
            Events.REPORT_SHARED,
          ]),
        ),
      )
      .groupBy(userEvents.userId);

    let sent = 0;
    let skipped = 0;

    for (const row of activity) {
      // Identity is resolved server-side from the contacts row — the userId
      // on events is the external id, never an email address.
      const contact = await db.query.contacts.findFirst({
        where: eq(contacts.externalId, row.userId),
      });
      if (!contact?.email) {
        skipped++;
        continue;
      }

      const result = await emailService.send({
        template: Templates.RETENTION_WEEKLY_DIGEST, // "retention-weekly-digest"
        to: contact.email,
        userId: row.userId,
        subject: "Your week in review",
        props: {
          reportsCreated: Number(row.reportsCreated),
          reportsShared: Number(row.reportsShared),
          weekOf: weekKey,
        },
        // NO skipPreferenceCheck — a digest is exactly the mail that
        // preferences exist to control. Unsubscribed/suppressed recipients
        // come back as a status, not an error.
        idempotencyKey: `digest:${row.userId}:${weekKey}`,
      });

      if (result.status === "sent") sent++;
      else skipped++;
    }

    logger.info("weekly-digest complete", { sent, skipped, weekKey });
    return { sent, skipped, week: weekKey };
  },
});

// src/workflows/index.ts — list only YOUR tasks; built-ins register themselves
import { weeklyDigestTask } from "./weekly-digest.js";

export const extraWorkflows = [weeklyDigestTask];

// src/worker.ts
import { createWorker } from "@hogsend/engine";
import { getContainer } from "./container.js";
import { journeys } from "./journeys/index.js";
import { extraWorkflows } from "./workflows/index.js";

const client = getContainer();
const worker = createWorker({
  container: client,
  journeys,
  extraWorkflows, // NOT `workflows`
});
await worker.start();