Design a Distributed Message Queue (Kafka)

import * as fs from "fs";
import * as path from "path";

const INDEX_INTERVAL_BYTES = 4096;

export class Segment {
  private fd: number;
  private indexFd: number;
  private bytesSinceIndex = 0;
  public bytesWritten = 0;

  constructor(dir: string, public baseOffset: number, public nextOffset: number) {
    const name = String(baseOffset).padStart(20, "0");
    this.fd = fs.openSync(path.join(dir, name + ".log"), "a");
    this.indexFd = fs.openSync(path.join(dir, name + ".index"), "a");
  }

  append(key: string | null, value: string, ts = Date.now()): number {
    const offset = this.nextOffset++;
    const payload = Buffer.from(JSON.stringify({ offset, ts, key, value }));
    const frame = Buffer.alloc(4 + payload.length);
    frame.writeUInt32BE(payload.length, 0);
    payload.copy(frame, 4);

    if (this.bytesSinceIndex >= INDEX_INTERVAL_BYTES) {
      const entry = Buffer.alloc(12);
      entry.writeUInt32BE(offset - this.baseOffset, 0); // relative offset
      entry.writeBigUInt64BE(BigInt(this.bytesWritten), 4); // byte position
      fs.writeSync(this.indexFd, entry);
      this.bytesSinceIndex = 0;
    }
    fs.writeSync(this.fd, frame);
    this.bytesWritten += frame.length;
    this.bytesSinceIndex += frame.length;
    return offset;
  }

  flush() { fs.fsyncSync(this.fd); } // called on a 10ms timer: group commit
}

import * as fs from "fs";

interface IndexEntry { relOffset: number; pos: number }

export function loadIndex(indexPath: string): IndexEntry[] {
  const buf = fs.readFileSync(indexPath);
  const entries: IndexEntry[] = [];
  for (let i = 0; i + 12 <= buf.length; i += 12) {
    entries.push({ relOffset: buf.readUInt32BE(i), pos: Number(buf.readBigUInt64BE(i + 4)) });
  }
  return entries;
}

// Binary search the sparse index for the greatest entry <= target,
// then scan forward frame by frame to the exact offset.
export function fetch(logPath: string, index: IndexEntry[], baseOffset: number,
                      targetOffset: number, maxRecords: number) {
  const rel = targetOffset - baseOffset;
  let lo = 0, hi = index.length - 1, startPos = 0;
  while (lo <= hi) {
    const mid = (lo + hi) >> 1;
    if (index[mid].relOffset <= rel) { startPos = index[mid].pos; lo = mid + 1; }
    else hi = mid - 1;
  }
  const buf = fs.readFileSync(logPath); // real impl: bounded pread, not whole file
  const out = [];
  let pos = startPos;
  while (pos + 4 <= buf.length && out.length < maxRecords) {
    const len = buf.readUInt32BE(pos);
    const rec = JSON.parse(buf.subarray(pos + 4, pos + 4 + len).toString());
    if (rec.offset >= targetOffset) out.push(rec);
    pos += 4 + len;
  }
  return out;
}

CREATE TABLE group_offsets (
  group_id   TEXT NOT NULL,
  topic      TEXT NOT NULL,
  partition  INTEGER NOT NULL,
  committed  BIGINT NOT NULL,       -- next offset to read
  generation INTEGER NOT NULL,      -- bumped on every rebalance
  updated_at TEXT NOT NULL DEFAULT (datetime('now')),
  PRIMARY KEY (group_id, topic, partition)
);

-- Commit is fenced: a consumer from an old generation (kicked out by a
-- rebalance it has not noticed yet) cannot clobber the new owner's progress.
UPDATE group_offsets
SET committed = :offset, generation = :gen, updated_at = datetime('now')
WHERE group_id = :group AND topic = :topic AND partition = :partition
  AND generation <= :gen
  AND committed < :offset;          -- offsets only move forward

-- Resume point after restart or rebalance:
SELECT partition, committed FROM group_offsets
WHERE group_id = :group AND topic = :topic;