Pub-sub message queue

interface Message { offset: number; value: string; }

/** An append-only log. Reading never removes anything: each consumer group just remembers how far it got. */
class Topic {
  readonly log: Message[] = [];
  constructor(readonly name: string) {}
  append(value: string): number {
    this.log.push({ offset: this.log.length, value });
    return this.log.length - 1;
  }
}

class Broker {
  private topics = new Map<string, Topic>();
  private committed = new Map<string, number>(); // "group|topic" -> next offset to read

  createTopic(name: string): void { this.topics.set(name, new Topic(name)); }
  publish(topic: string, value: string): number { return this.topics.get(topic)!.append(value); }

  /** Returns messages after the group's committed offset. Doesn't advance it: the consumer commits after processing. */
  poll(group: string, topic: string, max: number): Message[] {
    const start = this.committed.get(`${group}|${topic}`) ?? 0;
    return this.topics.get(topic)!.log.slice(start, start + max);
  }

  commit(group: string, topic: string, nextOffset: number): void {
    const k = `${group}|${topic}`;
    this.committed.set(k, Math.max(nextOffset, this.committed.get(k) ?? 0));
  }

  lag(group: string, topic: string): number {
    return this.topics.get(topic)!.log.length - (this.committed.get(`${group}|${topic}`) ?? 0);
  }
}

const values = (ms: Message[]) => ms.map((m) => `${m.offset}:${m.value}`).join(" ") || "(none)";

const b = new Broker();
b.createTopic("orders");
for (const v of ["o-100", "o-101", "o-102", "o-103"]) b.publish("orders", v);
let got = b.poll("billing", "orders", 2);
console.log("billing polls 2:", values(got));
b.commit("billing", "orders", got[got.length - 1].offset + 1);
console.log("email polls 3:", values(b.poll("email", "orders", 3)));
console.log("email crashes before committing; after restart it polls again");
got = b.poll("email", "orders", 3);
console.log("email polls 3:", values(got));
b.commit("email", "orders", got[got.length - 1].offset + 1);
console.log("billing polls 10:", values(b.poll("billing", "orders", 10)));
console.log("lag: billing", b.lag("billing", "orders"), "email", b.lag("email", "orders"));

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

record Message(int offset, String value) {}

/** An append-only log. Reading never removes anything: each consumer group just remembers how far it got. */
class Topic {
    final String name;
    final List<Message> log = new ArrayList<>();

    Topic(String name) {
        this.name = name;
    }

    int append(String value) {
        log.add(new Message(log.size(), value));
        return log.size() - 1;
    }
}

class Broker {
    private final Map<String, Topic> topics = new HashMap<>();
    private final Map<String, Integer> committed = new HashMap<>(); // "group|topic" -> next offset to read

    synchronized void createTopic(String name) {
        topics.put(name, new Topic(name));
    }

    synchronized int publish(String topic, String value) {
        return topics.get(topic).append(value);
    }

    /** Returns messages after the group's committed offset. Doesn't advance it: the consumer commits after processing. */
    synchronized List<Message> poll(String group, String topic, int max) {
        List<Message> log = topics.get(topic).log;
        int start = committed.getOrDefault(group + "|" + topic, 0);
        return List.copyOf(log.subList(start, Math.min(log.size(), start + max)));
    }

    synchronized void commit(String group, String topic, int nextOffset) {
        committed.merge(group + "|" + topic, nextOffset, Math::max);
    }

    synchronized int lag(String group, String topic) {
        return topics.get(topic).log.size() - committed.getOrDefault(group + "|" + topic, 0);
    }
}

public class Main {
    static String values(List<Message> ms) {
        return ms.isEmpty() ? "(none)" : ms.stream().map(m -> m.offset() + ":" + m.value()).collect(Collectors.joining(" "));
    }

    public static void main(String[] args) {
        Broker b = new Broker();
        b.createTopic("orders");
        for (String v : List.of("o-100", "o-101", "o-102", "o-103")) b.publish("orders", v);
        List<Message> got = b.poll("billing", "orders", 2);
        System.out.println("billing polls 2: " + values(got));
        b.commit("billing", "orders", got.get(got.size() - 1).offset() + 1);
        System.out.println("email polls 3: " + values(b.poll("email", "orders", 3)));
        System.out.println("email crashes before committing; after restart it polls again");
        got = b.poll("email", "orders", 3);
        System.out.println("email polls 3: " + values(got));
        b.commit("email", "orders", got.get(got.size() - 1).offset() + 1);
        System.out.println("billing polls 10: " + values(b.poll("billing", "orders", 10)));
        System.out.println("lag: billing " + b.lag("billing", "orders") + " email " + b.lag("email", "orders"));
    }
}