Denormalisation and Fan-Out

// Fan-out on write, with the trigger for the fan-out committed atomically
// alongside the post itself.
async function createPost(db, authorId, body) {
  return db.tx(async (t) => {
    const post = await t.one(
      `INSERT INTO posts (author_id, body)
       VALUES ($1, $2) RETURNING id, created_at`,
      [authorId, body],
    );

    // The outbox row commits with the post. If the worker is down, the row is
    // still here; if the transaction rolls back, there is no orphan event.
    // This is the only way to get "both or neither" across two systems.
    await t.none(
      `INSERT INTO outbox (topic, payload) VALUES ($1, $2)`,
      ['post.created', { postId: post.id, authorId, createdAt: post.created_at }],
    );
    return post;
  });
}

const FANOUT_FOLLOWER_LIMIT = 100_000;

// Hybrid fan-out: materialise into followers' timelines for ordinary accounts,
// and leave very-high-follower accounts to be merged in at read time.
async function handlePostCreated(db, { postId, authorId, createdAt }) {
  // Destructuring in the parameter list; the longhand is
  //   async function handlePostCreated(db, event) { const postId = event.postId; ... }
  const { followers } = await db.one(
    'SELECT followers FROM user_stats WHERE user_id = $1',
    [authorId],
  );

  if (followers > FANOUT_FOLLOWER_LIMIT) return; // readers merge this author live

  // One statement, server-side: do not pull a million follower ids into Node.
  // ON CONFLICT makes the worker safe to retry after a partial failure.
  // created_at is the post's own timestamp from the event, never the worker's
  // clock: the read path merges these rows with live ones by created_at, so a
  // retried or delayed fan-out would otherwise order the post as if it were new.
  await db.none(
    `INSERT INTO timeline (user_id, post_id, created_at)
     SELECT f.follower_id, $1, $2 FROM follows f WHERE f.followee_id = $3
     ON CONFLICT (user_id, post_id) DO NOTHING`,
    [postId, createdAt, authorId],
  );
}

// The read side pays for the hybrid: materialised rows, plus a live query for
// the handful of accounts that were skipped on write.
async function readTimeline(db, userId, limit = 50) {
  const [materialised, live] = await Promise.all([
    db.manyOrNone(
      `SELECT post_id, created_at FROM timeline
       WHERE user_id = $1 ORDER BY created_at DESC LIMIT $2`,
      [userId, limit],
    ),
    db.manyOrNone(
      `SELECT p.id AS post_id, p.created_at FROM posts p
       JOIN follows f ON f.followee_id = p.author_id
       JOIN user_stats s ON s.user_id = p.author_id
       WHERE f.follower_id = $1 AND s.followers > $2
       ORDER BY p.created_at DESC LIMIT $3`,
      [userId, FANOUT_FOLLOWER_LIMIT, limit],
    ),
  ]);

  return [...materialised, ...live]
    .sort((a, b) => b.created_at - a.created_at)
    .slice(0, limit);
}