feat: add auto theme, Redis failover, weekly digest, and campaign map

backend/src/config/redis.js

@@ -0,0 +1,92 @@
+// @ts-check
+import Redis from 'ioredis';
+
+/**
+ * Build a Redis client supporting standalone, Sentinel, and Cluster topologies.
+ *
+ * Environment variables:
+ * - REDIS_URL / REDIS_HOST  — standalone connection string
+ * - REDIS_SENTINEL_HOSTS    — comma-separated host:port sentinel endpoints
+ * - REDIS_SENTINEL_NAME     — sentinel master name (required with SENTINEL_HOSTS)
+ * - REDIS_SENTINEL_PASSWORD — optional sentinel auth password
+ * - REDIS_CLUSTER_NODES     — comma-separated host:port cluster seed nodes
+ * - REDIS_PASSWORD          — optional Redis auth password (all modes)
+ *
+ * Sentinel and Cluster modes give automatic failover when a node fails;
+ * ioredis reconnects to the promoted primary without application changes.
+ *
+ * @param {object} [options]
+ * @param {string} [options.redisUrl]        Override REDIS_URL
+ * @param {string} [options.sentinelHosts]   Override REDIS_SENTINEL_HOSTS
+ * @param {string} [options.sentinelName]    Override REDIS_SENTINEL_NAME
+ * @param {string} [options.sentinelPassword] Override REDIS_SENTINEL_PASSWORD
+ * @param {string} [options.clusterNodes]    Override REDIS_CLUSTER_NODES
+ * @param {string} [options.password]        Override REDIS_PASSWORD
+ * @param {object} [options.commonOptions]   Extra ioredis options
+ * @returns {{ mode: 'sentinel'|'cluster'|'standalone'|'disabled', client: import('ioredis').Redis | import('ioredis').Redis.Cluster | null }}
+ */
+export function createRedisClient(options = {}) {
+  const sentinelHosts = options.sentinelHosts ?? process.env.REDIS_SENTINEL_HOSTS;
+  const clusterNodes = options.clusterNodes ?? process.env.REDIS_CLUSTER_NODES;
+  const password = options.password ?? process.env.REDIS_PASSWORD;
+  const commonOptions = {
+    maxRetriesPerRequest: 3,
+    enableReadyCheck: true,
+    lazyConnect: false,
+    ...(password ? { password } : {}),
+    ...options.commonOptions,
+  };
+
+  if (sentinelHosts) {
+    const name = options.sentinelName ?? process.env.REDIS_SENTINEL_NAME;
+    if (!name) {
+      throw new Error('REDIS_SENTINEL_NAME is required when REDIS_SENTINEL_HOSTS is set');
+    }
+    const sentinels = parseHostList(sentinelHosts);
+    const sentinelPassword =
+      options.sentinelPassword ?? process.env.REDIS_SENTINEL_PASSWORD ?? undefined;
+    const client = new Redis({
+      ...commonOptions,
+      sentinels,
+      name,
+      ...(sentinelPassword ? { sentinelPassword } : {}),
+    });
+    return { mode: 'sentinel', client };
+  }
+
+  if (clusterNodes) {
+    const nodes = parseHostList(clusterNodes);
+    const client = new Redis.Cluster(nodes, {
+      redisOptions: commonOptions,
+      clusterRetryStrategy: (times) => Math.min(times * 200, 5_000),
+    });
+    return { mode: 'cluster', client };
+  }
+
+  const redisUrl = options.redisUrl ?? process.env.REDIS_URL ?? process.env.REDIS_HOST;
+  if (!redisUrl) {
+    return { mode: 'disabled', client: null };
+  }
+
+  const client = new Redis(redisUrl, commonOptions);
+  return { mode: 'standalone', client };
+}
+
+/**
+ * Parse a comma-separated "host:port" list into ioredis node descriptors.
+ * @param {string} value
+ * @returns {{ host: string, port: number }[]}
+ */
+export function parseHostList(value) {
+  return String(value)
+    .split(',')
+    .map((entry) => entry.trim())
+    .filter(Boolean)
+    .map((entry) => {
+      const [host, port] = entry.split(':');
+      if (!host || !port) {
+        throw new Error(`Invalid Redis host entry "${entry}" — expected host:port`);
+      }
+      return { host, port: Number(port) };
+    });
+}

backend/src/config/redis.test.js

@@ -0,0 +1,80 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { createRedisClient, parseHostList } from './redis.js';
+
+test('parseHostList: parses comma-separated host:port entries', () => {
+  const nodes = parseHostList('a.example:26379, b.example:26380');
+  assert.deepEqual(nodes, [
+    { host: 'a.example', port: 26379 },
+    { host: 'b.example', port: 26380 },
+  ]);
+});
+
+test('parseHostList: rejects entries missing host or port', () => {
+  assert.throws(() => parseHostList('nocolon'), /Invalid Redis host entry/);
+  assert.throws(() => parseHostList(':6379'), /Invalid Redis host entry/);
+});
+
+test('createRedisClient: returns disabled when no config is present', () => {
+  const saved = {
+    REDIS_URL: process.env.REDIS_URL,
+    REDIS_HOST: process.env.REDIS_HOST,
+    REDIS_SENTINEL_HOSTS: process.env.REDIS_SENTINEL_HOSTS,
+    REDIS_CLUSTER_NODES: process.env.REDIS_CLUSTER_NODES,
+  };
+  delete process.env.REDIS_URL;
+  delete process.env.REDIS_HOST;
+  delete process.env.REDIS_SENTINEL_HOSTS;
+  delete process.env.REDIS_CLUSTER_NODES;
+  try {
+    const result = createRedisClient();
+    assert.equal(result.client, null);
+    assert.equal(result.mode, 'disabled');
+  } finally {
+    for (const [key, value] of Object.entries(saved)) {
+      if (value === undefined) delete process.env[key];
+      else process.env[key] = value;
+    }
+  }
+});
+
+test('createRedisClient: throws when sentinel hosts are set without a master name', () => {
+  assert.throws(
+    () =>
+      createRedisClient({
+        sentinelHosts: 's1:26379',
+        sentinelName: undefined,
+      }),
+    /REDIS_SENTINEL_NAME is required/,
+  );
+});
+
+test('createRedisClient: builds a Sentinel client when hosts and name are provided', () => {
+  const { mode, client } = createRedisClient({
+    sentinelHosts: 's1:26379,s2:26379',
+    sentinelName: 'trivela-master',
+  });
+  assert.equal(mode, 'sentinel');
+  assert.ok(client, 'client must be created');
+  assert.equal(typeof client.connect, 'function');
+  client.disconnect();
+});
+
+test('createRedisClient: builds a Cluster client when cluster nodes are provided', () => {
+  const { mode, client } = createRedisClient({
+    clusterNodes: 'c1:6379,c2:6379',
+  });
+  assert.equal(mode, 'cluster');
+  assert.ok(client, 'client must be created');
+  assert.equal(typeof client.connect, 'function');
+  client.disconnect();
+});
+
+test('createRedisClient: builds a standalone client for a URL', () => {
+  const { mode, client } = createRedisClient({
+    redisUrl: 'redis://localhost:6379',
+  });
+  assert.equal(mode, 'standalone');
+  assert.ok(client, 'client must be created');
+  client.disconnect();
+});

backend/src/index.js

@@ -16,7 +16,7 @@ import compression from 'compression';
 import multer from 'multer';
 import { join } from 'node:path';
 import { pathToFileURL } from 'node:url';
-import Redis from 'ioredis';
+import { createRedisClient } from './config/redis.js';
 import createApiKeyAuth, { createMasterKeyAuth, readProvidedKey } from './middleware/apiKeyAuth.js';
 import { createRateLimiter, createRedisStore } from './middleware/rateLimit.js';
 import { createAuthLockout } from './middleware/authLockout.js';
@@ -96,6 +96,8 @@ import { createIdempotencyMiddleware } from './middleware/idempotency.js';
 import { createDistributedLock, createInMemoryLock } from './jobs/distributedLock.js';
 import { createExportJob } from './jobs/exportJob.js';
 import { createEventIndexer } from './jobs/eventIndexer.js';
+import { createWeeklyDigestJob } from './jobs/weeklyDigestJob.js';
+import { createEmailService } from './services/emailService.js';
 import { createSqliteJobQueueRepository } from './dal/sqliteJobQueueRepository.js';
 import { createDurableJobQueue } from './jobs/durableJobQueue.js';
 import {
@@ -581,23 +583,17 @@ export async function createApp(options = {}) {
 
   let rateLimitStore = null;
   let usageRedisClient = null;
-  const redisUrl = process.env.REDIS_URL || process.env.REDIS_HOST;
-  if (redisUrl && !options.disableRedis) {
+  if (!options.disableRedis) {
     try {
-      const redisClient = new Redis(redisUrl, {
-        maxRetriesPerRequest: 3,
-        enableReadyCheck: true,
-        lazyConnect: false,
-      });
-      redisClient.on('error', (err) => {
-        log.error({ err }, 'Redis connection error');
-      });
-      rateLimitStore = createRedisStore(redisClient);
-      usageRedisClient = redisClient;
-      log.info(
-        { redisUrl: redisUrl.replace(/:[^:@]+@/, ':***@') },
-        'Rate limiter using Redis store',
-      );
+      const { mode, client: redisClient } = createRedisClient();
+      if (redisClient) {
+        redisClient.on('error', (err) => {
+          log.error({ err }, 'Redis connection error');
+        });
+        rateLimitStore = createRedisStore(redisClient);
+        usageRedisClient = redisClient;
+        log.info({ redisMode: mode }, 'Rate limiter using Redis store');
+      }
     } catch (error) {
       log.warn(
         { err: error },
@@ -617,6 +613,28 @@ export async function createApp(options = {}) {
       ? createDistributedLock(usageRedisClient, { ttlMs: lockTtlMs })
       : createInMemoryLock({ ttlMs: lockTtlMs }));
 
+  // Email service adapter — wraps sendEmail with .send({to, template, data}) interface
+  const emailService = {
+    async send({ to, template, data }) {
+      const { sendEmail } = await import('./services/emailService.js');
+      return sendEmail(template, to, data);
+    },
+  };
+
+  // Weekly campaign digest — performance summary emails for operators (#1355)
+  const digestRecipients = (process.env.CAMPAIGN_DIGEST_RECIPIENTS ?? '')
+    .split(',')
+    .map((s) => s.trim())
+    .filter(Boolean);
+  const weeklyDigestJob = createWeeklyDigestJob({
+    db: dal.db,
+    emailService,
+    campaignRepository,
+    referralRepository,
+    getRecipients: () => digestRecipients,
+    logger: log,
+  });
+
   // Data export job — daily CSV export to object storage (#562)
   const exportRetentionDays = normalizePositiveInteger(
     /** @type {any} */ (options.exportRetentionDays) ?? process.env.EXPORT_RETENTION_DAYS,
@@ -886,6 +904,9 @@ export async function createApp(options = {}) {
       async campaign_reward_notifications() {
         await campaignRewardNotificationJob();
       },
+      async weekly_digest() {
+        await weeklyDigestJob.run();
+      },
     },
     logger: log,
     deadLetter: failedJobRepository,
@@ -945,6 +966,13 @@ export async function createApp(options = {}) {
     ).unref?.();
   }
 
+  // Weekly campaign digest — fires on startup (idempotent per ISO week) and weekly (#1355)
+  if (!options.disableJobs) {
+    const runDigest = () => jobRunner.enqueue('weekly_digest', null);
+    runDigest();
+    setInterval(runDigest, 7 * 24 * 60 * 60 * 1_000).unref?.();
+  }
+
   // #1255 — IPFS pinning of campaign assets (enabled when PINATA_JWT is set)
   const ipfsPinService =
     /** @type {any} */ (options.ipfsPinService) ??