Weekly digest

// 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();

// src/container.ts — lazy so importing a workflow module stays side-effect free
import { createHogsendClient, type HogsendClient } from "@hogsend/engine";
import { templates } from "./emails/registry.js";
import { journeys } from "./journeys/index.js";

let client: HogsendClient | undefined;

export function getContainer(): HogsendClient {
  client ??= createHogsendClient({ journeys, email: { templates } });
  return client;
}