feat: live class sessions, CE workflow, and docs (#683)

Adds live-class session scheduling on lessons (join link, date/time,
timezone) with automatic 24h and 1h reminder emails plus a calendar
invite (.ics), a time-change notification with an updated invite, a
teacher-authored welcome email and session invite on enrollment, and a
"Notify students" quiz nudge.

LMS gets an upcoming live-sessions card and Join buttons, plus a
reusable live-session card component (3 states) in @cio/ui with a
Storybook story. The sidebar content tree now reflects sequential
progression locking like the content list, and the course-completion
confetti regression is fixed.

Org-based emails no longer reference ClassroomIO by name. Docs add
live-class, welcome-email, and course-progression guides, expand the
certificates guide (design, signatures, corrected labels), move
plan-required callouts to the top, flatten how-to-guides to the docs
root with redirects, and add screenshots. Includes an add-docs-image
skill.

apps/jobs/src/index.ts

@@ -3,11 +3,12 @@ import './bootstrap';
 import './workers/media';
 import './workers/media-transcribe';
 import './workers/emails';
+import './workers/notifications';
 import './workers/maintenance';
 import './workers/agent-course-generation';
 
 import { log } from './utils/logger';
 
 log.info('all-workers-running', {
-  workers: ['media', 'media-transcribe', 'emails', 'maintenance', 'agent-course-generation']
+  workers: ['media', 'media-transcribe', 'emails', 'notifications', 'maintenance', 'agent-course-generation']
 });

apps/jobs/src/processors/emails/send.ts

@@ -22,7 +22,8 @@ export async function processSendEmail(rawPayload: unknown): Promise<SendResult>
       // generic record was already validated in the API helper before enqueue.
       fields: payload.fields as never,
       from: payload.from,
-      replyTo: payload.replyTo
+      replyTo: payload.replyTo,
+      ics: payload.ics
     });
     const result = extractProviderId(responses);
     log.info('email-sent', {

apps/jobs/src/processors/notifications/index.ts

@@ -0,0 +1,3 @@
+export * from './notify-course-exercise';
+export * from './session-reminder-scan';
+export * from './notify-session-update';

apps/jobs/src/processors/notifications/notify-course-exercise.ts

@@ -0,0 +1,56 @@
+import { getCourseMembers } from '@cio/db/queries/course/people';
+import { ZNotifyCourseExercisePayload, enqueueEmailSend } from '@cio/jobs';
+import { ROLE } from '@cio/utils/constants';
+
+import { log } from '../../utils/logger';
+
+interface NotifyResult {
+  notified: number;
+}
+
+/**
+ * Process a `notifications:notify-course-exercise` job. Fans out one
+ * `quizAssigned` email per course member that has an email, each carrying a
+ * deep link to take the quiz. Returns how many emails were enqueued (BullMQ
+ * stores this as the job return value for the dashboard poll).
+ */
+export async function processNotifyCourseExercise(rawPayload: unknown, jobId: string): Promise<NotifyResult> {
+  const payload = ZNotifyCourseExercisePayload.parse(rawPayload);
+
+  const members = await getCourseMembers(payload.courseId);
+  const recipients = members
+    .filter((member) => member.roleId === ROLE.STUDENT)
+    .map((member) => member.profile?.email)
+    .filter((email): email is string => !!email);
+
+  let notified = 0;
+  for (const email of recipients) {
+    try {
+      await enqueueEmailSend(
+        {
+          kind: 'template',
+          template: 'quizAssigned',
+          to: email,
+          fields: {
+            orgName: payload.orgName,
+            courseName: payload.courseName,
+            exerciseTitle: payload.exerciseTitle,
+            quizUrl: payload.quizUrl,
+            branding: payload.branding
+          },
+          from: payload.fromName
+        },
+        // Vary by the notify job id so re-running the nudge re-sends.
+        { idempotencyKey: `quiz-assigned:${jobId}:${email}` }
+      );
+      notified += 1;
+    } catch (error) {
+      log.error('notify-course-exercise-enqueue-failed', { jobId, recipient: email });
+      throw error;
+    }
+  }
+
+  log.info('notify-course-exercise-done', { jobId, courseId: payload.courseId, notified });
+
+  return { notified };
+}

apps/jobs/src/processors/notifications/notify-session-update.ts

@@ -0,0 +1,104 @@
+import { getCourseById, getCourseWithOrgData } from '@cio/db/queries/course';
+import { getCourseMembers } from '@cio/db/queries/course/people';
+import { getLessonById } from '@cio/db/queries/lesson';
+import { ROLE } from '@cio/utils/constants';
+import { buildEmailBranding, buildEmailFromName, buildSessionIcs } from '@cio/email';
+import { ZNotifyCourseSessionUpdatePayload, enqueueEmailSend } from '@cio/jobs';
+
+import { log } from '../../utils/logger';
+
+interface UpdateResult {
+  notified: number;
+}
+
+function formatSessionTime(lessonAt: string, timezone: string | null): string {
+  try {
+    return new Intl.DateTimeFormat('en-US', {
+      timeZone: timezone ?? 'UTC',
+      dateStyle: 'medium',
+      timeStyle: 'short',
+      timeZoneName: 'short'
+    }).format(new Date(lessonAt));
+  } catch {
+    return new Date(lessonAt).toUTCString();
+  }
+}
+
+/**
+ * Emails every course member an UPDATED calendar invite after a teacher changes
+ * a live session's time/link. Same UID + a higher SEQUENCE (derived from the
+ * lesson's updatedAt) with METHOD:REQUEST so calendars update the existing event.
+ */
+export async function processNotifyCourseSessionUpdate(rawPayload: unknown): Promise<UpdateResult> {
+  const { courseId, lessonId } = ZNotifyCourseSessionUpdatePayload.parse(rawPayload);
+
+  const [courseRows, orgData, lesson] = await Promise.all([
+    getCourseById(courseId),
+    getCourseWithOrgData(courseId),
+    getLessonById(lessonId)
+  ]);
+
+  const course = courseRows?.[0];
+  if (!course || !orgData || !lesson || !lesson.lessonAt || !lesson.callUrl) {
+    log.warn('notify-session-update-skipped', { courseId, lessonId, reason: 'missing session data' });
+    return { notified: 0 };
+  }
+
+  const sessionTimezone = course.metadata?.sessionTimezone ?? null;
+  const orgName = orgData.orgName || 'ClassroomIO';
+  const branding = buildEmailBranding({
+    name: orgData.orgName,
+    avatarUrl: orgData.orgAvatarUrl,
+    theme: orgData.orgTheme
+  });
+  const sessionTimeLabel = formatSessionTime(lesson.lessonAt, sessionTimezone);
+  const sequence = Math.floor(Date.parse(lesson.updatedAt ?? new Date().toISOString()) / 1000);
+
+  const ics = buildSessionIcs({
+    uid: `session-${lesson.id}@classroomio`,
+    sequence,
+    method: 'REQUEST',
+    start: lesson.lessonAt,
+    title: lesson.title,
+    description: `Join your live session: ${lesson.callUrl}`,
+    url: lesson.callUrl,
+    alarmsBeforeMinutes: [1440, 60]
+  });
+
+  const members = await getCourseMembers(courseId);
+  const recipients = members
+    .filter((member) => member.roleId === ROLE.STUDENT)
+    .map((member) => member.profile?.email)
+    .filter((email): email is string => !!email);
+
+  let notified = 0;
+  for (const email of recipients) {
+    try {
+      await enqueueEmailSend(
+        {
+          kind: 'template',
+          template: 'sessionUpdated',
+          to: email,
+          fields: {
+            orgName,
+            courseName: course.title,
+            sessionTitle: lesson.title,
+            sessionTimeLabel,
+            joinUrl: lesson.callUrl,
+            branding
+          },
+          from: buildEmailFromName(`${orgName} (via ClassroomIO.com)`),
+          ics
+        },
+        { idempotencyKey: `session-updated:${lesson.id}:${email}:${sequence}` }
+      );
+      notified += 1;
+    } catch (error) {
+      log.error('notify-session-update-enqueue-failed', { lessonId, recipient: email });
+    }
+  }
+
+  log.info('notify-session-update-done', { courseId, lessonId, notified });
+
+  return { notified };
+}

apps/jobs/src/processors/notifications/session-reminder-scan.ts

@@ -0,0 +1,102 @@
+import { listUpcomingSessionsForReminderScan } from '@cio/db/queries/course';
+import { buildEmailBranding, buildEmailFromName, buildSessionIcs } from '@cio/email';
+import { enqueueEmailSend } from '@cio/jobs';
+
+import { log } from '../../utils/logger';
+
+interface ScanResult {
+  scanned: number;
+  remindersEnqueued: number;
+}
+
+const OFFSETS_MINUTES = [1440, 60] as const;
+
+function formatSessionTime(lessonAt: string, timezone: string | null): string {
+  try {
+    return new Intl.DateTimeFormat('en-US', {
+      timeZone: timezone ?? 'UTC',
+      dateStyle: 'medium',
+      timeStyle: 'short',
+      timeZoneName: 'short'
+    }).format(new Date(lessonAt));
+  } catch {
+    return new Date(lessonAt).toUTCString();
+  }
+}
+
+/**
+ * Scan upcoming live sessions and enqueue 24h / 1h reminder emails to enrolled
+ * students. Idempotency key per (lesson, student, offset) so each reminder
+ * fires once (BullMQ dedupes on jobId). The 24h reminder also carries an .ics.
+ */
+export async function processSessionReminderScan(): Promise<ScanResult> {
+  const rows = await listUpcomingSessionsForReminderScan();
+  const now = Date.now();
+
+  let remindersEnqueued = 0;
+
+  for (const row of rows) {
+    if (!row.email) continue;
+
+    const startMs = new Date(row.lessonAt).getTime();
+    if (Number.isNaN(startMs)) continue;
+
+    const minutesUntil = Math.round((startMs - now) / 60_000);
+    if (minutesUntil <= 0) continue;
+
+    const branding = buildEmailBranding({
+      name: row.organizationName,
+      avatarUrl: row.organizationAvatarUrl,
+      theme: row.organizationTheme
+    });
+    const sessionTimeLabel = formatSessionTime(row.lessonAt, row.sessionTimezone);
+
+    for (const offset of OFFSETS_MINUTES) {
+      if (minutesUntil > offset) continue;
+
+      const whenLabel = offset === 1440 ? 'in about 24 hours' : 'in about 1 hour';
+      const ics =
+        offset === 1440
+          ? buildSessionIcs({
+              uid: `session-${row.lessonId}@classroomio`,
+              sequence: 0,
+              method: 'PUBLISH',
+              start: row.lessonAt,
+              title: row.lessonTitle,
+              description: `Join your live session: ${row.callUrl}`,
+              url: row.callUrl,
+              alarmsBeforeMinutes: [1440, 60]
+            })
+          : undefined;
+
+      try {
+        await enqueueEmailSend(
+          {
+            kind: 'template',
+            template: 'sessionReminder',
+            to: row.email,
+            fields: {
+              orgName: row.organizationName,
+              courseName: row.courseName,
+              sessionTitle: row.lessonTitle,
+              sessionTimeLabel,
+              whenLabel,
+              joinUrl: row.callUrl,
+              branding
+            },
+            from: buildEmailFromName(`${row.organizationName} (via ClassroomIO.com)`),
+            ...(ics ? { ics } : {})
+          },
+          { idempotencyKey: `session-reminder:${row.lessonId}:${row.profileId}:${offset}` }
+        );
+        remindersEnqueued += 1;
+      } catch (error) {
+        log.error('session-reminder-enqueue-failed', { lessonId: row.lessonId, offset });
+      }
+    }
+  }
+
+  log.info('session-reminder-scan-done', { scanned: rows.length, remindersEnqueued });
+
+  return { scanned: rows.length, remindersEnqueued };
+}

apps/jobs/src/workers/notifications.ts

@@ -0,0 +1,102 @@
+import './../bootstrap';
+
+import { Queue, Worker } from 'bullmq';
+
+import { recordDeadLetterJob } from '@cio/db/queries';
+import { JOB_NAMES, QUEUE_NAMES, createRedisConnection } from '@cio/jobs';
+
+import { errorMessage } from '../utils/cancel';
+import { env } from '../config/env';
+import { log } from '../utils/logger';
+import {
+  processNotifyCourseExercise,
+  processNotifyCourseSessionUpdate,
+  processSessionReminderScan
+} from '../processors/notifications';
+
+const concurrency = Number.parseInt(env.EMAIL_WORKER_CONCURRENCY ?? '5', 10) || 5;
+const connection = createRedisConnection();
+
+const worker = new Worker(
+  QUEUE_NAMES.notifications,
+  async (job) => {
+    log.info('notification-job-start', {
+      jobName: job.name,
+      bullmqJobId: job.id,
+      attempt: job.attemptsMade + 1
+    });
+
+    switch (job.name) {
+      case JOB_NAMES.notifications.notifyCourseExercise:
+        return processNotifyCourseExercise(job.data, String(job.id));
+      case JOB_NAMES.notifications.sessionReminderScan:
+        return processSessionReminderScan();
+      case JOB_NAMES.notifications.notifyCourseSessionUpdate:
+        return processNotifyCourseSessionUpdate(job.data);
+      default:
+        throw new Error(`Unknown notifications job: ${job.name}`);
+    }
+  },
+  { connection, concurrency }
+);
+
+// Repeatable scheduler: run the session-reminder scan every 15 minutes.
+const SESSION_REMINDER_SCAN_INTERVAL_MS = 15 * 60 * 1_000;
+const schedulerConnection = createRedisConnection();
+const notificationsQueue = new Queue(QUEUE_NAMES.notifications, { connection: schedulerConnection });
+
+async function registerSchedulers(): Promise<void> {
+  try {
+    await notificationsQueue.upsertJobScheduler(
+      'session-reminder-scan-scheduler',
+      { every: SESSION_REMINDER_SCAN_INTERVAL_MS },
+      { name: JOB_NAMES.notifications.sessionReminderScan, data: {} }
+    );
+    log.info('session-reminder-scan-scheduler-registered', { everyMs: SESSION_REMINDER_SCAN_INTERVAL_MS });
+  } catch (err) {
+    log.error('session-reminder-scan-scheduler-register-failed', { error: errorMessage(err) });
+  }
+}
+
+void registerSchedulers();
+
+worker.on('failed', async (job, err) => {
+  if (!job) return;
+
+  log.error('notification-job-failed', {
+    jobName: job.name,
+    bullmqJobId: job.id,
+    attempt: job.attemptsMade,
+    error: errorMessage(err)
+  });
+
+  const isFinalAttempt = job.attemptsMade >= (job.opts.attempts ?? 1);
+  if (!isFinalAttempt) return;
+
+  await recordDeadLetterJob({
+    organizationId: null,
+    domain: 'notifications',
+    runId: null,
+    queueName: QUEUE_NAMES.notifications,
+    jobName: job.name,
+    bullmqJobId: job.id ?? null,
+    payload: job.data as Record<string, unknown>,
+    error: { code: 'WORKER_EXHAUSTED_RETRIES', message: errorMessage(err), stack: err.stack },
+    attempts: job.attemptsMade
+  });
+});
+
+worker.on('ready', () => log.info('notifications-worker-ready'));
+worker.on('error', (err) => log.error('notifications-worker-error', { error: errorMessage(err) }));
+
+const shutdown = async (signal: string) => {
+  log.info('notifications-worker-shutdown', { signal });
+  await worker.close();
+  await notificationsQueue.close();
+  await schedulerConnection.quit();
+  await connection.quit();
+  process.exit(0);
+};
+
+process.on('SIGTERM', () => void shutdown('SIGTERM'));
+process.on('SIGINT', () => void shutdown('SIGINT'));