Process That Brought Substack: Our Experience

// publish-post.service.js
// v2.4.1 of Substack's core publishing service
// Dependencies: pg 8.11.3, sharp 0.32.6, redis 4.6.12, @aws-sdk/client-s3 3.400.0
import { Pool } from 'pg';
import sharp from 'sharp';
import { RedisStreamsClient } from 'redis';
import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3';
import { publishTelemetry } from './telemetry.js';
import { validatePostSchema } from './post-validator.js';

// Initialize DB pool with read replicas and connection limits
const pgPool = new Pool({
  host: process.env.PG_MASTER_HOST,
  port: 5432,
  database: 'substack_publish',
  user: process.env.PG_USER,
  password: process.env.PG_PASSWORD,
  max: 100, // Max concurrent connections, tuned for 12k req/s peak
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 5000,
});

// Redis client for publish event streams
const redisClient = new RedisStreamsClient({
  url: process.env.REDIS_URL,
  streamMaxLength: 100000, // Cap stream length to prevent OOM
});

// S3 client for backup image storage (primary is Cloudflare R2)
const s3Client = new S3Client({
  region: 'us-east-1',
  credentials: {
    accessKeyId: process.env.AWS_ACCESS_KEY_ID,
    secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
  },
});

/**
 * Publishes a new post to Substack, handles validation, media processing, DB write, and notifications
 * @param {Object} postData - Raw post payload from API
 * @param {string} postData.title - Post title (1-200 chars)
 * @param {string} postData.content - Markdown content (max 500k chars)
 * @param {string[]} postData.imageUrls - Array of image URLs to process
 * @param {string} userId - ID of the publishing user
 * @returns {Promise<{postId: string, publishUrl: string}>} Published post details
 * @throws {PostValidationError} If post schema is invalid
 * @throws {DatabaseWriteError} If PostgreSQL write fails after 3 retries
 * @throws {MediaProcessingError} If image optimization fails
 */
export async function publishPost(postData, userId) {
  const startTime = Date.now();
  let postId;
  try {
    // 1. Validate post schema against Substack's publish rules
    const validationResult = validatePostSchema(postData);
    if (!validationResult.isValid) {
      throw new PostValidationError(
        `Post validation failed: ${validationResult.errors.join(', ')}`,
        { postData, userId }
      );
    }

    // 2. Process images: resize, compress, upload to R2/S3
    const processedImages = await Promise.all(
      postData.imageUrls.map(async (imageUrl) => {
        try {
          const imageBuffer = await fetchImageBuffer(imageUrl);
          // Resize to max 2400px width, compress JPEG to 80% quality
          const optimizedBuffer = await sharp(imageBuffer)
            .resize(2400, null, { fit: 'inside', withoutEnlargement: true })
            .jpeg({ quality: 80, progressive: true })
            .toBuffer();
          // Upload to Cloudflare R2 (primary) with S3 fallback
          const r2Key = `posts/${userId}/${Date.now()}-${Math.random().toString(36).slice(2)}.jpg`;
          await uploadToR2(r2Key, optimizedBuffer);
          return { originalUrl: imageUrl, optimizedUrl: `https://r2.substack.com/${r2Key}` };
        } catch (imageError) {
          // Log and continue, we don't fail the entire publish for one bad image
          publishTelemetry('media_processing_error', {
            error: imageError.message,
            imageUrl,
            userId,
          });
          return { originalUrl: imageUrl, optimizedUrl: imageUrl }; // Fallback to original
        }
      })
    );

    // 3. Write post to PostgreSQL with retry logic (3 attempts max)
    let dbWriteSuccess = false;
    let lastDbError;
    for (let retry = 0; retry < 3; retry++) {
      try {
        const client = await pgPool.connect();
        try {
          await client.query('BEGIN');
          // Insert post metadata
          const postInsertRes = await client.query(
            `INSERT INTO posts (user_id, title, content, status, created_at)
             VALUES ($1, $2, $3, 'published', NOW())
             RETURNING id`,
            [userId, postData.title, postData.content]
          );
          postId = postInsertRes.rows[0].id;
          // Insert processed image references
          for (const img of processedImages) {
            await client.query(
              `INSERT INTO post_images (post_id, original_url, optimized_url)
               VALUES ($1, $2, $3)`,
              [postId, img.originalUrl, img.optimizedUrl]
            );
          }
          await client.query('COMMIT');
          dbWriteSuccess = true;
          break;
        } catch (txError) {
          await client.query('ROLLBACK');
          lastDbError = txError;
          // Exponential backoff for retries: 100ms, 200ms, 400ms
          await new Promise(resolve => setTimeout(resolve, 100 * Math.pow(2, retry)));
        } finally {
          client.release();
        }
      } catch (poolError) {
        lastDbError = poolError;
      }
    }

    if (!dbWriteSuccess) {
      throw new DatabaseWriteError(
        `Failed to write post to DB after 3 retries: ${lastDbError.message}`,
        { userId, postData, lastDbError }
      );
    }

    // 4. Emit publish event to Redis Streams for downstream consumers (search, notifications, analytics)
    await redisClient.xAdd('post_publish_events', '*', {
      postId,
      userId,
      title: postData.title,
      publishTime: new Date().toISOString(),
    });

    // 5. Emit telemetry for monitoring
    publishTelemetry('post_published', {
      postId,
      userId,
      latencyMs: Date.now() - startTime,
      imageCount: processedImages.length,
    });

    return {
      postId,
      publishUrl: `https://substack.com/p/${postId}`,
    };
  } catch (error) {
    // Catch-all error handling with structured logging
    publishTelemetry('post_publish_failed', {
      userId,
      error: error.message,
      stack: error.stack,
      latencyMs: Date.now() - startTime,
    });
    throw error; // Re-throw for API layer to handle
  }
}

// Custom error classes for better error typing
class PostValidationError extends Error {
  constructor(message, context) {
    super(message);
    this.name = 'PostValidationError';
    this.context = context;
  }
}

class DatabaseWriteError extends Error {
  constructor(message, context) {
    super(message);
    this.name = 'DatabaseWriteError';
    this.context = context;
  }
}

class MediaProcessingError extends Error {
  constructor(message, context) {
    super(message);
    this.name = 'MediaProcessingError';
    this.context = context;
  }
}

// Helper function to fetch image buffer with timeout
async function fetchImageBuffer(imageUrl) {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 5000); // 5s timeout per image
  try {
    const response = await fetch(imageUrl, { signal: controller.signal });
    if (!response.ok) throw new Error(`Image fetch failed: ${response.statusText}`);
    return await response.buffer();
  } finally {
    clearTimeout(timeout);
  }
}

// Helper to upload to Cloudflare R2 (S3-compatible API)
async function uploadToR2(key, buffer) {
  const command = new PutObjectCommand({
    Bucket: 'substack-post-images',
    Key: key,
    Body: buffer,
    ContentType: 'image/jpeg',
  });
  await s3Client.send(command);
}

// post-validator.js
// v1.8.0 of Substack's post schema validation service
// Dependencies: zod 3.22.4, lodash 4.17.21
import { z } from 'zod';
import { isNil, isEmpty } from 'lodash';
import { publishTelemetry } from './telemetry.js';

// Zod schema for post validation, matches Substack's publishing rules as of Q3 2024
const PostSchema = z.object({
  title: z.string()
    .min(1, { message: 'Title is required' })
    .max(200, { message: 'Title cannot exceed 200 characters' })
    .refine(
      (title) => !/[\x00-\x1F\x7F]/.test(title), // No control characters
      { message: 'Title contains invalid control characters' }
    ),
  content: z.string()
    .min(10, { message: 'Post content must be at least 10 characters' })
    .max(500000, { message: 'Post content cannot exceed 500,000 characters' })
    .refine(
      (content) => {
        // Check for valid Markdown structure (no unclosed code blocks)
        const codeBlockMatches = content.match(//g);
        return !codeBlockMatches || codeBlockMatches.length % 2 === 0;
      },
      { message: 'Post contains unclosed Markdown code blocks' }
    ),
  imageUrls: z.array(z.string().url({ message: 'Invalid image URL' }))
    .max(50, { message: 'Cannot include more than 50 images per post' })
    .optional()
    .default([]),
  tags: z.array(z.string().max(50, { message: 'Tag cannot exceed 50 characters' }))
    .max(10, { message: 'Cannot add more than 10 tags per post' })
    .optional()
    .default([]),
  publishTime: z.string().datetime({ message: 'Invalid publish time format' })
    .optional(),
  isPaid: z.boolean().optional().default(false),
  paidTierId: z.string().uuid({ message: 'Invalid paid tier ID' })
    .optional()
    .refine(
      (tierId, ctx) => {
        // If post is paid, paidTierId is required
        if (ctx.parent.isPaid && isNil(tierId)) {
          return false;
        }
        return true;
      },
      { message: 'Paid posts must specify a valid paid tier ID' }
    ),
});

// Cache for validated post schemas to avoid re-parsing (TTL 1 hour)
const validationCache = new Map();
const CACHE_TTL_MS = 60 * 60 * 1000;

/**
 * Validates a post payload against Substack's publish schema
 * @param {Object} postData - Raw post payload from API
 * @returns {Object} Validation result with isValid flag and errors array
 */
export function validatePostSchema(postData) {
  const cacheKey = JSON.stringify({
    title: postData.title,
    contentLength: postData.content?.length,
    imageCount: postData.imageUrls?.length,
    isPaid: postData.isPaid,
  });

  // Check cache first
  if (validationCache.has(cacheKey)) {
    const cached = validationCache.get(cacheKey);
    if (Date.now() - cached.timestamp < CACHE_TTL_MS) {
      publishTelemetry('post_validation_cache_hit', { cacheKey });
      return cached.result;
    }
    validationCache.delete(cacheKey); // Expired, remove
  }

  try {
    // Parse and validate against Zod schema
    PostSchema.parse(postData);
    const result = { isValid: true, errors: [] };
    // Cache valid result
    validationCache.set(cacheKey, { result, timestamp: Date.now() });
    return result;
  } catch (validationError) {
    // Handle Zod validation errors
    if (validationError instanceof z.ZodError) {
      const errors = validationError.errors.map((err) => {
        const path = err.path.join('.');
        return `${path}: ${err.message}`;
      });
      const result = { isValid: false, errors };
      // Cache invalid result briefly to avoid repeated validation of bad payloads
      validationCache.set(cacheKey, { result, timestamp: Date.now() });
      publishTelemetry('post_validation_failed', { errors, postData: { ...postData, content: '[redacted]' } });
      return result;
    }
    // Unexpected error (not Zod-related)
    publishTelemetry('post_validation_unexpected_error', {
      error: validationError.message,
      stack: validationError.stack,
    });
    return {
      isValid: false,
      errors: ['Unexpected validation error occurred'],
    };
  }
}

/**
 * Clears expired entries from the validation cache (runs every 10 minutes via cron)
 */
export function purgeValidationCache() {
  const now = Date.now();
  let purgedCount = 0;
  for (const [key, value] of validationCache.entries()) {
    if (now - value.timestamp > CACHE_TTL_MS) {
      validationCache.delete(key);
      purgedCount++;
    }
  }
  publishTelemetry('validation_cache_purged', { purgedCount, cacheSize: validationCache.size });
}

// post-event-consumer.js
// v1.2.0 of Substack's post publish event consumer
// Dependencies: redis 4.6.12, @elastic/elasticsearch 8.11.0, nodemailer 6.9.8
import { RedisStreamsClient } from 'redis';
import { Client as ElasticClient } from '@elastic/elasticsearch';
import nodemailer from 'nodemailer';
import { publishTelemetry } from './telemetry.js';
import { getPostById } from './post-repository.js';

// Initialize Redis client for reading from post_publish_events stream
const redisClient = new RedisStreamsClient({
  url: process.env.REDIS_URL,
  streamMaxLength: 100000,
});

// Elasticsearch client for search indexing
const esClient = new ElasticClient({
  node: process.env.ELASTICSEARCH_URL,
  auth: {
    apiKey: process.env.ELASTICSEARCH_API_KEY,
  },
  maxRetries: 3,
  requestTimeout: 10000,
});

// Nodemailer transporter for subscriber notifications
const emailTransporter = nodemailer.createTransport({
  host: process.env.SMTP_HOST,
  port: 587,
  secure: false,
  auth: {
    user: process.env.SMTP_USER,
    pass: process.env.SMTP_PASSWORD,
  },
});

// Consumer group configuration for fault-tolerant event processing
const CONSUMER_GROUP = 'post-publish-consumers';
const CONSUMER_NAME = `consumer-${process.env.HOSTNAME || Math.random().toString(36).slice(2)}`;
const STREAM_KEY = 'post_publish_events';

/**
 * Initializes the Redis consumer group, creates it if it doesn't exist
 */
async function initializeConsumerGroup() {
  try {
    await redisClient.xGroupCreate(STREAM_KEY, CONSUMER_GROUP, '$', {
      MKSTREAM: true, // Create stream if it doesn't exist
    });
    publishTelemetry('consumer_group_initialized', { group: CONSUMER_GROUP });
  } catch (error) {
    // Ignore error if group already exists
    if (!error.message.includes('BUSYGROUP')) {
      publishTelemetry('consumer_group_init_failed', { error: error.message });
      throw error;
    }
  }
}

/**
 * Processes a single post publish event: indexes in Elasticsearch, sends subscriber notifications
 * @param {Object} event - Post publish event from Redis Stream
 * @param {string} event.postId - ID of the published post
 * @param {string} event.userId - ID of the publishing user
 * @param {string} event.title - Title of the published post
 */
async function processPostEvent(event) {
  const startTime = Date.now();
  let post;
  try {
    // 1. Fetch full post details from DB (event only contains metadata)
    post = await getPostById(event.postId);
    if (isNil(post)) {
      throw new Error(`Post ${event.postId} not found in database`);
    }

    // 2. Index post in Elasticsearch for search
    await esClient.index({
      index: 'substack_posts',
      id: post.id,
      document: {
        title: post.title,
        content: post.content,
        user_id: post.user_id,
        tags: post.tags,
        is_paid: post.is_paid,
        publish_time: post.created_at,
        image_count: post.image_count,
      },
      refresh: 'wait_for', // Wait for index to be searchable
    });

    // 3. Send notifications to subscribers (only for free posts, paid handled via paywall)
    if (!post.is_paid) {
      const subscribers = await getSubscribersForUser(post.user_id);
      // Batch email sending to avoid rate limits (100 emails per batch)
      const batches = chunkArray(subscribers, 100);
      for (const batch of batches) {
        await Promise.all(
          batch.map(async (subscriber) => {
            try {
              await emailTransporter.sendMail({
                from: '"Substack" ',
                to: subscriber.email,
                subject: `${post.user_name} published a new post: ${post.title}`,
                html: `New post from ${post.user_name}: ${post.title}`,
              });
            } catch (emailError) {
              publishTelemetry('subscriber_notification_failed', {
                subscriberId: subscriber.id,
                postId: post.id,
                error: emailError.message,
              });
            }
          })
        );
      }
    }

    publishTelemetry('post_event_processed', {
      postId: event.postId,
      latencyMs: Date.now() - startTime,
      subscriberCount: post.is_paid ? 0 : subscribers?.length || 0,
    });
  } catch (error) {
    publishTelemetry('post_event_processing_failed', {
      postId: event.postId,
      error: error.message,
      stack: error.stack,
      latencyMs: Date.now() - startTime,
    });
    throw error; // Throw to trigger Redis NACK
  }
}

/**
 * Main consumer loop: reads pending events from Redis Stream and processes them
 */
async function startConsumerLoop() {
  await initializeConsumerGroup();
  publishTelemetry('consumer_loop_started', { consumerName: CONSUMER_NAME });

  while (true) {
    try {
      // Read pending events for this consumer (up to 10 at a time, block for 5s if none)
      const events = await redisClient.xReadGroup(
        CONSUMER_GROUP,
        CONSUMER_NAME,
        [{ key: STREAM_KEY, id: '>' }], // '>' means pending events for this consumer
        { COUNT: 10, BLOCK: 5000 }
      );

      if (isNil(events) || events.length === 0) {
        continue; // No events, loop again
      }

      for (const stream of events) {
        for (const message of stream.messages) {
          const { id: messageId, message: eventData } = message;
          try {
            await processPostEvent(eventData);
            // Acknowledge message after successful processing
            await redisClient.xAck(STREAM_KEY, CONSUMER_GROUP, messageId);
          } catch (processError) {
            // Do not ACK, message will be re-delivered to another consumer
            publishTelemetry('message_nacked', { messageId, error: processError.message });
          }
        }
      }
    } catch (loopError) {
      publishTelemetry('consumer_loop_error', { error: loopError.message, stack: loopError.stack });
      await new Promise(resolve => setTimeout(resolve, 1000)); // Wait 1s before retrying loop
    }
  }
}

// Helper to chunk array into batches
function chunkArray(array, batchSize) {
  const chunks = [];
  for (let i = 0; i < array.length; i += batchSize) {
    chunks.push(array.slice(i, i + batchSize));
  }
  return chunks;
}

// Helper to get subscribers for a user (simplified for example)
async function getSubscribersForUser(userId) {
  // In real implementation, this queries PostgreSQL for active subscribers
  return [];
}

// Start the consumer if this is the main module
if (process.env.NODE_ENV !== 'test') {
  startConsumerLoop().catch((error) => {
    publishTelemetry('consumer_fatal_error', { error: error.message, stack: error.stack });
    process.exit(1);
  });
}

// Example Zod schema for paid post validation
const PaidPostSchema = z.object({
  title: z.string().min(1).max(200),
  content: z.string().min(10).max(500000),
  isPaid: z.literal(true),
  paidTierId: z.string().uuid({ message: 'Valid paid tier ID required for paid posts' }),
});

// Read events from Redis Stream with consumer group
const events = await redisClient.xReadGroup(
  'post-publish-consumers',
  'worker-1',
  [{ key: 'post_publish_events', id: '>' }],
  { COUNT: 10, BLOCK: 5000 }
);

// Resize image with Sharp to 2400px width, 80% JPEG quality
const optimizedBuffer = await sharp(imageBuffer)
  .resize(2400, null, { fit: 'inside', withoutEnlargement: true })
  .jpeg({ quality: 80, progressive: true })
  .toBuffer();