Low-level design

Oct 2026

Comments and Replies - In-Memory Solution, Ordering & Tests

Full CommentService Python+TS sketch with tests for order, validation, isolation, independent IDs, concurrency, snapshots.

1Gist
2Maps
3Q&A
4Sandbox

View

Voice readout needs Web Speech Synthesis in this browser.

Why it exists: The spec is believable only when a test can mutate a returned thread and watch the store stay intact, and when ten threads cannot mint the same id.
Definition: CommentService is the in-memory implementation: frozen views, two counters, a list per post, a dict by comment id, one RLock, and seven tests for order, validation, isolation, snapshots, and concurrent ids.
30-second answer: Validate before taking the lock where you can. Under the lock, reject a missing post or comment, bump the matching counter and the order counter, append, and return a new frozen view. get_thread deep-copies. The concurrent test runs 10 threads times 50 comments and checks unique sorted ids.
Mental model: Every mutation that publishes an id also publishes the append, under the same lock. The object you return is a value, not a handle.
Key tradeoff: Deep copy on get_thread costs an allocation per read. Sharing the internal list is faster and incorrect the moment a caller appends.
Interview cue: Read add_comment and add_reply side by side and show the counters differ. Then point at the snapshot test and the concurrent test as the proof.

Question ladder

L1
What does test_order_independent_of_clock assert?
Answer
Five comments listed in the order they were added, and that sequence is sorted because ids are monotonic.
L2
What does an empty author raise?
Answer
CommentError INVALID_AUTHOR. A blank body is INVALID_BODY. A missing post is POST_NOT_FOUND.
L3
How do independent ids show up in numbers?
Answer
The first comment is 1, its first reply is 1, and the next comment is 2. Reply ids do not consume comment ids.
L4
What is post isolation?
Answer
A comment on p1 never appears in list_comments for p2. Each post has its own list.
L5
How does the snapshot test work?
Answer
It mutates the returned dict and appends a fake reply. A second get_thread still has the original body and an empty reply list.
L6
What does the concurrent test check?
Answer
10 threads times 50 comments, no errors, 500 unique ids, and ids equal to their sorted order.
L7
What does the TypeScript sketch omit?
Answer
The lock, validation, and deep copy. It shows the same maps and counters for an interview transfer, not a second implementation to maintain.

Failure modes

Mutated snapshot
Writing into a returned thread dict must not change the next get_thread. deepcopy and frozen views are the fix.
Shared counter
One sequence for comments and replies fails test_independent_id_sequences.
Duplicate concurrent ids
A counter bump outside the lock fails the uniqueness assertion.

Misconceptions

Frozen CommentView means the thread dict is safe too.
get_thread builds dicts. Those need deepcopy. The frozen view only protects list_comments.
The TypeScript class is thread-safe because the shape matches.
The sketch has no lock. The Python service is the one under test.
List order can be fixed later by sorting ids.
Ids happen to be monotonic here because the lock couples bump and append. Say that. Do not sort by time.

Interviewer traps

Rewrite the service in a framework mid-interview.
Stay on the class the tests import.
Drop the snapshot test as obvious.
It is the test that catches a returned internal list.

Design scenario

Same prompt for every reader.

Requirements

Two levels, independent counters, creation order, validation, isolation, snapshots.

Traffic / scale

One process. The stress test is 500 comments from 10 threads.

Latency

Memory operations under one RLock.

Consistency

Ids unique and monotonic. List order equals append order. Posts do not leak into each other.

Availability

Bad input raises CommentError with a stable code. It does not insert a partial row.

Failure assumptions

Callers mutate returned objects.
Threads share one CommentService.

Constraints

Do not share one counter across comments and replies.
Do not return parent.replies directly.

Prompt

Implement CommentService so the seven tests pass, including concurrent unique ids and a defensive thread snapshot.

API

Which CommentError codes do the tests assert?

Data

Which two collections are updated when a comment is added?

Architecture

Where is the lock acquired relative to validation?

What the tests are protecting

Prefer

Frozen views and a deep-copied thread

The store changes only inside CommentService methods.

list_comments builds new CommentView values.
get_thread runs deepcopy.
The mutation test would fail if either were a live reference.

Alternative

Return the internal list

Faster, and the caller's append becomes your data.

A reply list grows without add_reply.
Ids and order stop matching the service.
The snapshot test is there to catch it.

Try this

Run the seven tests. Then make list_comments return the internal objects and watch the snapshot or isolation assertions fail.

Overview

Full Python solution for the two-level Comments and Replies service plus unittest coverage for order, validation, independent IDs, post isolation, defensive snapshots and concurrent unique ID allocation. A TypeScript sketch of the same structure sits at the end for interview transfer.

Diagrams

Add under the lock, return a copy

Diagram 1 of the write path the tests cover. A missing parent is NOT_FOUND. A success appends and returns a snapshot.

1
Validate author and body
Empty, non-string, or over-long input raises CommentError before the structure changes.
2
Acquire the RLock
Counter bumps and appends share this critical section.
3
Post or parent exists?
add_comment needs the post. add_reply needs the comment id in the index.
4
NOT_FOUND
POST_NOT_FOUND or COMMENT_NOT_FOUND. No id is consumed.
5
Bump the matching sequence
comment_seq and reply_seq move independently. order_seq stamps creation order.
6
Append and return a snapshot
The list keeps order. The caller receives a frozen view or a deep copy, not the internal object.

Zoom

Decisions

1
1. Validate author and body
- next2. Acquire RLock
2
2. Acquire RLock
- next3. Post or parent exists?
?
3. Post or parent exists?
- No4. Failure: NOT_FOUND
- Yes5. Bump comment or reply seq
4
4. Failure: NOT_FOUND
5
5. Bump comment or reply seq
- next6. Append in creation order
6
6. Append in creation order
- next7. Return a frozen snapshot
7
7. Return a frozen snapshot

Lesson map

Comments and Replies - In-Memory Solution, Ordering & Tests

Full CommentService Python+TS sketch with tests for order, validation, isolation, independent IDs, concurrency, snapshots.

1. Validate author and body2. Acquire RLock3. Post or parent exists?4. Failure: NOT_FOUND5. Bump comment or reply seq6. Append in creation order7. Return a frozen snapshot

Select a node to see why it exists, or an edge to see the protocol, direction, effect, and consequence.

Mermaid export

flowchart TB
  a["1. Validate author and body"]
  b["2. Acquire RLock"]
  c["3. Post or parent exists?"]
  d["4. Failure: NOT_FOUND"]
  e["5. Bump comment or reply seq"]
  f["6. Append in creation order"]
  g["7. Return a frozen snapshot"]
  a -->|continues| b
  b -->|continues| c
  c -->|No| d
  c -->|Yes| e
  e -->|continues| f
  f -->|continues| g

Zoom

Decisions

?
Need O1 attach and order?
- YesDict by id plus list per post
- Order onlyList alone is a slow reply path
2
Dict by id plus list per post
- nextOne lock covers both indexes
3
List alone is a slow reply path
- nextOne lock covers both indexes
4
One lock covers both indexes

Solution (Python)

python

"""Comments and Replies: in-memory two-level service.
Sandbox: python3 service.py
Tests: python3 test_service.py
"""
from __future__ import annotations

import copy
import threading
from dataclasses import dataclass, field
from typing import Any, Optional


class CommentError(Exception):
    def __init__(self, code: str, message: str = "") -> None:
        self.code = code
        super().__init__(message or code)


@dataclass(frozen=True)
class CommentView:
    id: int
    post_id: str
    author: str
    body: str
    created_seq: int


@dataclass(frozen=True)
class ReplyView:
    id: int
    comment_id: int
    post_id: str
    author: str
    body: str
    created_seq: int


@dataclass
class _Comment:
    id: int
    post_id: str
    author: str
    body: str
    created_seq: int
    replies: list[_Reply] = field(default_factory=list)


@dataclass
class _Reply:
    id: int
    comment_id: int
    post_id: str
    author: str
    body: str
    created_seq: int


class CommentService:
    """Two-level comments (post -> comment -> reply). No deeper nesting.

    Design choices:
    - Per-post dict of ordered comment lists preserves creation order by append.
    - Independent auto-increment counters for comments and replies (not shared).
    - Global RLock for counter + structure mutations under concurrency.
    - list_* returns frozen dataclasses (snapshots); callers cannot mutate internals.
    """

    MAX_AUTHOR = 64
    MAX_BODY = 2000

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._comment_seq = 0
        self._reply_seq = 0
        self._order_seq = 0
        # post_id -> list[_Comment] in creation order
        self._posts: dict[str, list[_Comment]] = {}
        # comment_id -> _Comment for O(1) reply attach
        self._comments_by_id: dict[int, _Comment] = {}

    def _validate_author_body(self, author: str, body: str) -> None:
        if not isinstance(author, str) or not author.strip() or len(author) > self.MAX_AUTHOR:
            raise CommentError("INVALID_AUTHOR")
        if not isinstance(body, str) or not body.strip() or len(body) > self.MAX_BODY:
            raise CommentError("INVALID_BODY")

    def _validate_post_id(self, post_id: str) -> None:
        if not isinstance(post_id, str) or not post_id.strip() or len(post_id) > 128:
            raise CommentError("INVALID_POST_ID")

    def ensure_post(self, post_id: str) -> None:
        """Register an empty post container (idempotent)."""
        self._validate_post_id(post_id)
        with self._lock:
            self._posts.setdefault(post_id, [])

    def add_comment(self, post_id: str, author: str, body: str) -> CommentView:
        self._validate_post_id(post_id)
        self._validate_author_body(author, body)
        with self._lock:
            if post_id not in self._posts:
                raise CommentError("POST_NOT_FOUND")
            self._comment_seq += 1
            self._order_seq += 1
            c = _Comment(
                id=self._comment_seq,
                post_id=post_id,
                author=author.strip(),
                body=body.strip(),
                created_seq=self._order_seq,
            )
            self._posts[post_id].append(c)
            self._comments_by_id[c.id] = c
            return CommentView(c.id, c.post_id, c.author, c.body, c.created_seq)

    def add_reply(self, comment_id: int, author: str, body: str) -> ReplyView:
        if not isinstance(comment_id, int) or comment_id < 1:
            raise CommentError("INVALID_COMMENT_ID")
        self._validate_author_body(author, body)
        with self._lock:
            parent = self._comments_by_id.get(comment_id)
            if parent is None:
                raise CommentError("COMMENT_NOT_FOUND")
            self._reply_seq += 1
            self._order_seq += 1
            r = _Reply(
                id=self._reply_seq,
                comment_id=comment_id,
                post_id=parent.post_id,
                author=author.strip(),
                body=body.strip(),
                created_seq=self._order_seq,
            )
            parent.replies.append(r)
            return ReplyView(r.id, r.comment_id, r.post_id, r.author, r.body, r.created_seq)

    def list_comments(self, post_id: str) -> list[CommentView]:
        self._validate_post_id(post_id)
        with self._lock:
            if post_id not in self._posts:
                raise CommentError("POST_NOT_FOUND")
            # Snapshot: new list of frozen views in creation order.
            return [
                CommentView(c.id, c.post_id, c.author, c.body, c.created_seq)
                for c in self._posts[post_id]
            ]

    def list_replies(self, comment_id: int) -> list[ReplyView]:
        if not isinstance(comment_id, int) or comment_id < 1:
            raise CommentError("INVALID_COMMENT_ID")
        with self._lock:
            parent = self._comments_by_id.get(comment_id)
            if parent is None:
                raise CommentError("COMMENT_NOT_FOUND")
            return [
                ReplyView(r.id, r.comment_id, r.post_id, r.author, r.body, r.created_seq)
                for r in parent.replies
            ]

    def get_thread(self, post_id: str) -> list[dict[str, Any]]:
        """Defensive deep snapshot: comments with nested reply dicts."""
        self._validate_post_id(post_id)
        with self._lock:
            if post_id not in self._posts:
                raise CommentError("POST_NOT_FOUND")
            out = []
            for c in self._posts[post_id]:
                out.append(
                    {
                        "id": c.id,
                        "post_id": c.post_id,
                        "author": c.author,
                        "body": c.body,
                        "created_seq": c.created_seq,
                        "replies": [
                            {
                                "id": r.id,
                                "comment_id": r.comment_id,
                                "author": r.author,
                                "body": r.body,
                                "created_seq": r.created_seq,
                            }
                            for r in c.replies
                        ],
                    }
                )
            return copy.deepcopy(out)


if __name__ == "__main__":
    svc = CommentService()
    svc.ensure_post("post-1")
    c1 = svc.add_comment("post-1", "ada", "First")
    c2 = svc.add_comment("post-1", "bob", "Second")
    r1 = svc.add_reply(c1.id, "cy", "Reply to first")
    print("comments", [c.id for c in svc.list_comments("post-1")])
    print("replies_on_c1", [r.id for r in svc.list_replies(c1.id)])
    print("thread", svc.get_thread("post-1"))

Tests

python

from __future__ import annotations

import threading
import unittest

from service import CommentError, CommentService


class CommentServiceTests(unittest.TestCase):
    def setUp(self):
        self.svc = CommentService()
        self.svc.ensure_post("p1")

    def test_order_independent_of_clock(self):
        ids = []
        for i in range(5):
            ids.append(self.svc.add_comment("p1", "a", f"b{i}").id)
        listed = [c.id for c in self.svc.list_comments("p1")]
        self.assertEqual(listed, ids)
        self.assertEqual(listed, sorted(listed))

    def test_two_levels_only_and_relationship(self):
        c = self.svc.add_comment("p1", "a", "hi")
        r = self.svc.add_reply(c.id, "b", "yo")
        self.assertEqual(r.comment_id, c.id)
        self.assertEqual(r.post_id, "p1")
        with self.assertRaises(CommentError) as ctx:
            self.svc.add_reply(9999, "b", "nope")
        self.assertEqual(ctx.exception.code, "COMMENT_NOT_FOUND")

    def test_validation(self):
        with self.assertRaises(CommentError):
            self.svc.add_comment("p1", "", "x")
        with self.assertRaises(CommentError):
            self.svc.add_comment("p1", "a", " ")
        with self.assertRaises(CommentError):
            self.svc.add_comment("missing", "a", "x")

    def test_independent_id_sequences(self):
        c1 = self.svc.add_comment("p1", "a", "1")
        r1 = self.svc.add_reply(c1.id, "b", "r")
        c2 = self.svc.add_comment("p1", "a", "2")
        # comment ids and reply ids increment independently
        self.assertEqual(c1.id, 1)
        self.assertEqual(c2.id, 2)
        self.assertEqual(r1.id, 1)

    def test_post_isolation(self):
        self.svc.ensure_post("p2")
        self.svc.add_comment("p1", "a", "only-p1")
        self.svc.add_comment("p2", "a", "only-p2")
        self.assertEqual(len(self.svc.list_comments("p1")), 1)
        self.assertEqual(len(self.svc.list_comments("p2")), 1)
        self.assertEqual(self.svc.list_comments("p1")[0].body, "only-p1")

    def test_defensive_snapshot(self):
        self.svc.add_comment("p1", "a", "x")
        thread = self.svc.get_thread("p1")
        thread[0]["body"] = "MUTATED"
        thread[0]["replies"].append({"id": 99})
        again = self.svc.get_thread("p1")
        self.assertEqual(again[0]["body"], "x")
        self.assertEqual(again[0]["replies"], [])

    def test_concurrent_ids_unique(self):
        errors = []
        n_threads = 10
        per = 50

        def worker():
            try:
                for i in range(per):
                    self.svc.add_comment("p1", "a", f"c-{threading.get_ident()}-{i}")
            except Exception as e:
                errors.append(e)

        threads = [threading.Thread(target=worker) for _ in range(n_threads)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        self.assertEqual(errors, [])
        comments = self.svc.list_comments("p1")
        self.assertEqual(len(comments), n_threads * per)
        ids = [c.id for c in comments]
        self.assertEqual(len(ids), len(set(ids)))
        self.assertEqual(ids, sorted(ids))


if __name__ == "__main__":
    unittest.main(verbosity=2)

Test run: Ran 7 tests in 0.005s — OK.

TypeScript sketch

typescript

// Sandbox sketch: mirrors the Python structure (Map + arrays + mutex via async queue or mutexify).
type Comment = { id: number; postId: string; author: string; body: string; createdSeq: number; replies: Reply[] };
type Reply = { id: number; commentId: number; postId: string; author: string; body: string; createdSeq: number };

class CommentService {
  private commentSeq = 0;
  private replySeq = 0;
  private orderSeq = 0;
  private posts = new Map<string, Comment[]>();
  private byId = new Map<number, Comment>();

  ensurePost(postId: string) {
    if (!this.posts.has(postId)) this.posts.set(postId, []);
  }

  addComment(postId: string, author: string, body: string): Omit<Comment, "replies"> {
    const list = this.posts.get(postId);
    if (!list) throw new Error("POST_NOT_FOUND");
    this.commentSeq += 1; this.orderSeq += 1;
    const c: Comment = { id: this.commentSeq, postId, author, body, createdSeq: this.orderSeq, replies: [] };
    list.push(c); this.byId.set(c.id, c);
    return { id: c.id, postId, author, body, createdSeq: c.createdSeq };
  }

  addReply(commentId: number, author: string, body: string): Reply {
    const parent = this.byId.get(commentId);
    if (!parent) throw new Error("COMMENT_NOT_FOUND");
    this.replySeq += 1; this.orderSeq += 1;
    const r: Reply = { id: this.replySeq, commentId, postId: parent.postId, author, body, createdSeq: this.orderSeq };
    parent.replies.push(r);
    return { ...r };
  }

  listComments(postId: string) {
    const list = this.posts.get(postId);
    if (!list) throw new Error("POST_NOT_FOUND");
    return list.map(({ id, postId, author, body, createdSeq }) => ({ id, postId, author, body, createdSeq }));
  }
}

Concepts used, learn more

Read the underlying idea on its own study page. This lesson applies it. It does not replace those pages.

Interview Q&A

Why validate author length before the lock?

Show

Answer

A bad author should not wait on the lock and should not bump a counter. The missing-parent checks stay inside the lock because they read shared maps.

What does a reply store for post_id?

Show

Answer

The parent comment's post_id, copied at insert time, so the reply does not need a second lookup to prove which post it belongs to.

Which pages explain the lock and the interface style?

Show

Answer

The service boundary is interfaces, state, and tradeoffs. The RLock is mutexes, condition variables, and deadlocks. Read-heavy locking is mutex versus RWLock. Lock-free counters are atomics and memory ordering, and this solution does not use them.

What does test_two_levels_only_and_relationship check?

Show

Answer

A reply stores the parent comment id and that comment's post id. add_reply(9999) raises COMMENT_NOT_FOUND. There is no API that attaches a reply to a reply.

Why is get_thread a list of dicts instead of frozen views?

Show

Answer

The nested shape is easier to assert in tests and to sketch in TypeScript. deepcopy detaches the reply arrays. list_comments stays on frozen dataclasses because it is flat.

What would a shared comment and reply counter break?

Show

Answer

test_independent_id_sequences expects the first reply id to be 1 even after a comment id 1 exists. One counter would make that reply id 2.

Pitfalls

Mutating a returned thread dict and corrupting the store (fixed with deepcopy / frozen views).
Sharing one counter for comments and replies when the prompt asks for independent sequences.

The series pager also walks these pages.

Low-level design

Part 2 of 3 · Comments and Replies

Oct 2026

Comments and Replies - In-Memory Solution, Ordering & Tests

Full CommentService Python+TS sketch with tests for order, validation, isolation, independent IDs, concurrency, snapshots.

1Gist
2Maps
3Q&A
4Sandbox

View

Voice readout needs Web Speech Synthesis in this browser.

Why it exists: The spec is believable only when a test can mutate a returned thread and watch the store stay intact, and when ten threads cannot mint the same id.
Definition: CommentService is the in-memory implementation: frozen views, two counters, a list per post, a dict by comment id, one RLock, and seven tests for order, validation, isolation, snapshots, and concurrent ids.
30-second answer: Validate before taking the lock where you can. Under the lock, reject a missing post or comment, bump the matching counter and the order counter, append, and return a new frozen view. get_thread deep-copies. The concurrent test runs 10 threads times 50 comments and checks unique sorted ids.
Mental model: Every mutation that publishes an id also publishes the append, under the same lock. The object you return is a value, not a handle.
Key tradeoff: Deep copy on get_thread costs an allocation per read. Sharing the internal list is faster and incorrect the moment a caller appends.
Interview cue: Read add_comment and add_reply side by side and show the counters differ. Then point at the snapshot test and the concurrent test as the proof.

Question ladder

L1
What does test_order_independent_of_clock assert?
Answer
Five comments listed in the order they were added, and that sequence is sorted because ids are monotonic.
L2
What does an empty author raise?
Answer
CommentError INVALID_AUTHOR. A blank body is INVALID_BODY. A missing post is POST_NOT_FOUND.
L3
How do independent ids show up in numbers?
Answer
The first comment is 1, its first reply is 1, and the next comment is 2. Reply ids do not consume comment ids.
L4
What is post isolation?
Answer
A comment on p1 never appears in list_comments for p2. Each post has its own list.
L5
How does the snapshot test work?
Answer
It mutates the returned dict and appends a fake reply. A second get_thread still has the original body and an empty reply list.
L6
What does the concurrent test check?
Answer
10 threads times 50 comments, no errors, 500 unique ids, and ids equal to their sorted order.
L7
What does the TypeScript sketch omit?
Answer
The lock, validation, and deep copy. It shows the same maps and counters for an interview transfer, not a second implementation to maintain.

Failure modes

Mutated snapshot
Writing into a returned thread dict must not change the next get_thread. deepcopy and frozen views are the fix.
Shared counter
One sequence for comments and replies fails test_independent_id_sequences.
Duplicate concurrent ids
A counter bump outside the lock fails the uniqueness assertion.

Misconceptions

Frozen CommentView means the thread dict is safe too.
get_thread builds dicts. Those need deepcopy. The frozen view only protects list_comments.
The TypeScript class is thread-safe because the shape matches.
The sketch has no lock. The Python service is the one under test.
List order can be fixed later by sorting ids.
Ids happen to be monotonic here because the lock couples bump and append. Say that. Do not sort by time.

Interviewer traps

Rewrite the service in a framework mid-interview.
Stay on the class the tests import.
Drop the snapshot test as obvious.
It is the test that catches a returned internal list.

Design scenario

Same prompt for every reader.

Requirements

Two levels, independent counters, creation order, validation, isolation, snapshots.

Traffic / scale

One process. The stress test is 500 comments from 10 threads.

Latency

Memory operations under one RLock.

Consistency

Ids unique and monotonic. List order equals append order. Posts do not leak into each other.

Availability

Bad input raises CommentError with a stable code. It does not insert a partial row.

Failure assumptions

Callers mutate returned objects.
Threads share one CommentService.

Constraints

Do not share one counter across comments and replies.
Do not return parent.replies directly.

Prompt

Implement CommentService so the seven tests pass, including concurrent unique ids and a defensive thread snapshot.

API

Which CommentError codes do the tests assert?

Data

Which two collections are updated when a comment is added?

Architecture

Where is the lock acquired relative to validation?

What the tests are protecting

Prefer

Frozen views and a deep-copied thread

The store changes only inside CommentService methods.

list_comments builds new CommentView values.
get_thread runs deepcopy.
The mutation test would fail if either were a live reference.

Alternative

Return the internal list

Faster, and the caller's append becomes your data.

A reply list grows without add_reply.
Ids and order stop matching the service.
The snapshot test is there to catch it.

Try this

Run the seven tests. Then make list_comments return the internal objects and watch the snapshot or isolation assertions fail.

Overview

Diagrams

Add under the lock, return a copy

Diagram 1 of the write path the tests cover. A missing parent is NOT_FOUND. A success appends and returns a snapshot.

1
Validate author and body
Empty, non-string, or over-long input raises CommentError before the structure changes.
2
Acquire the RLock
Counter bumps and appends share this critical section.
3
Post or parent exists?
add_comment needs the post. add_reply needs the comment id in the index.
4
NOT_FOUND
POST_NOT_FOUND or COMMENT_NOT_FOUND. No id is consumed.
5
Bump the matching sequence
comment_seq and reply_seq move independently. order_seq stamps creation order.
6
Append and return a snapshot
The list keeps order. The caller receives a frozen view or a deep copy, not the internal object.

Zoom

Decisions

1
1. Validate author and body
- next2. Acquire RLock
2
2. Acquire RLock
- next3. Post or parent exists?
?
3. Post or parent exists?
- No4. Failure: NOT_FOUND
- Yes5. Bump comment or reply seq
4
4. Failure: NOT_FOUND
5
5. Bump comment or reply seq
- next6. Append in creation order
6
6. Append in creation order
- next7. Return a frozen snapshot
7
7. Return a frozen snapshot

Lesson map

Comments and Replies - In-Memory Solution, Ordering & Tests

Full CommentService Python+TS sketch with tests for order, validation, isolation, independent IDs, concurrency, snapshots.

1. Validate author and body2. Acquire RLock3. Post or parent exists?4. Failure: NOT_FOUND5. Bump comment or reply seq6. Append in creation order7. Return a frozen snapshot

Select a node to see why it exists, or an edge to see the protocol, direction, effect, and consequence.

Mermaid export

flowchart TB
  a["1. Validate author and body"]
  b["2. Acquire RLock"]
  c["3. Post or parent exists?"]
  d["4. Failure: NOT_FOUND"]
  e["5. Bump comment or reply seq"]
  f["6. Append in creation order"]
  g["7. Return a frozen snapshot"]
  a -->|continues| b
  b -->|continues| c
  c -->|No| d
  c -->|Yes| e
  e -->|continues| f
  f -->|continues| g

Zoom

Decisions

?
Need O1 attach and order?
- YesDict by id plus list per post
- Order onlyList alone is a slow reply path
2
Dict by id plus list per post
- nextOne lock covers both indexes
3
List alone is a slow reply path
- nextOne lock covers both indexes
4
One lock covers both indexes

Solution (Python)

python

"""Comments and Replies: in-memory two-level service.
Sandbox: python3 service.py
Tests: python3 test_service.py
"""
from __future__ import annotations

import copy
import threading
from dataclasses import dataclass, field
from typing import Any, Optional


class CommentError(Exception):
    def __init__(self, code: str, message: str = "") -> None:
        self.code = code
        super().__init__(message or code)


@dataclass(frozen=True)
class CommentView:
    id: int
    post_id: str
    author: str
    body: str
    created_seq: int


@dataclass(frozen=True)
class ReplyView:
    id: int
    comment_id: int
    post_id: str
    author: str
    body: str
    created_seq: int


@dataclass
class _Comment:
    id: int
    post_id: str
    author: str
    body: str
    created_seq: int
    replies: list[_Reply] = field(default_factory=list)


@dataclass
class _Reply:
    id: int
    comment_id: int
    post_id: str
    author: str
    body: str
    created_seq: int


class CommentService:
    """Two-level comments (post -> comment -> reply). No deeper nesting.

    Design choices:
    - Per-post dict of ordered comment lists preserves creation order by append.
    - Independent auto-increment counters for comments and replies (not shared).
    - Global RLock for counter + structure mutations under concurrency.
    - list_* returns frozen dataclasses (snapshots); callers cannot mutate internals.
    """

    MAX_AUTHOR = 64
    MAX_BODY = 2000

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._comment_seq = 0
        self._reply_seq = 0
        self._order_seq = 0
        # post_id -> list[_Comment] in creation order
        self._posts: dict[str, list[_Comment]] = {}
        # comment_id -> _Comment for O(1) reply attach
        self._comments_by_id: dict[int, _Comment] = {}

    def _validate_author_body(self, author: str, body: str) -> None:
        if not isinstance(author, str) or not author.strip() or len(author) > self.MAX_AUTHOR:
            raise CommentError("INVALID_AUTHOR")
        if not isinstance(body, str) or not body.strip() or len(body) > self.MAX_BODY:
            raise CommentError("INVALID_BODY")

    def _validate_post_id(self, post_id: str) -> None:
        if not isinstance(post_id, str) or not post_id.strip() or len(post_id) > 128:
            raise CommentError("INVALID_POST_ID")

    def ensure_post(self, post_id: str) -> None:
        """Register an empty post container (idempotent)."""
        self._validate_post_id(post_id)
        with self._lock:
            self._posts.setdefault(post_id, [])

    def add_comment(self, post_id: str, author: str, body: str) -> CommentView:
        self._validate_post_id(post_id)
        self._validate_author_body(author, body)
        with self._lock:
            if post_id not in self._posts:
                raise CommentError("POST_NOT_FOUND")
            self._comment_seq += 1
            self._order_seq += 1
            c = _Comment(
                id=self._comment_seq,
                post_id=post_id,
                author=author.strip(),
                body=body.strip(),
                created_seq=self._order_seq,
            )
            self._posts[post_id].append(c)
            self._comments_by_id[c.id] = c
            return CommentView(c.id, c.post_id, c.author, c.body, c.created_seq)

    def add_reply(self, comment_id: int, author: str, body: str) -> ReplyView:
        if not isinstance(comment_id, int) or comment_id < 1:
            raise CommentError("INVALID_COMMENT_ID")
        self._validate_author_body(author, body)
        with self._lock:
            parent = self._comments_by_id.get(comment_id)
            if parent is None:
                raise CommentError("COMMENT_NOT_FOUND")
            self._reply_seq += 1
            self._order_seq += 1
            r = _Reply(
                id=self._reply_seq,
                comment_id=comment_id,
                post_id=parent.post_id,
                author=author.strip(),
                body=body.strip(),
                created_seq=self._order_seq,
            )
            parent.replies.append(r)
            return ReplyView(r.id, r.comment_id, r.post_id, r.author, r.body, r.created_seq)

    def list_comments(self, post_id: str) -> list[CommentView]:
        self._validate_post_id(post_id)
        with self._lock:
            if post_id not in self._posts:
                raise CommentError("POST_NOT_FOUND")
            # Snapshot: new list of frozen views in creation order.
            return [
                CommentView(c.id, c.post_id, c.author, c.body, c.created_seq)
                for c in self._posts[post_id]
            ]

    def list_replies(self, comment_id: int) -> list[ReplyView]:
        if not isinstance(comment_id, int) or comment_id < 1:
            raise CommentError("INVALID_COMMENT_ID")
        with self._lock:
            parent = self._comments_by_id.get(comment_id)
            if parent is None:
                raise CommentError("COMMENT_NOT_FOUND")
            return [
                ReplyView(r.id, r.comment_id, r.post_id, r.author, r.body, r.created_seq)
                for r in parent.replies
            ]

    def get_thread(self, post_id: str) -> list[dict[str, Any]]:
        """Defensive deep snapshot: comments with nested reply dicts."""
        self._validate_post_id(post_id)
        with self._lock:
            if post_id not in self._posts:
                raise CommentError("POST_NOT_FOUND")
            out = []
            for c in self._posts[post_id]:
                out.append(
                    {
                        "id": c.id,
                        "post_id": c.post_id,
                        "author": c.author,
                        "body": c.body,
                        "created_seq": c.created_seq,
                        "replies": [
                            {
                                "id": r.id,
                                "comment_id": r.comment_id,
                                "author": r.author,
                                "body": r.body,
                                "created_seq": r.created_seq,
                            }
                            for r in c.replies
                        ],
                    }
                )
            return copy.deepcopy(out)


if __name__ == "__main__":
    svc = CommentService()
    svc.ensure_post("post-1")
    c1 = svc.add_comment("post-1", "ada", "First")
    c2 = svc.add_comment("post-1", "bob", "Second")
    r1 = svc.add_reply(c1.id, "cy", "Reply to first")
    print("comments", [c.id for c in svc.list_comments("post-1")])
    print("replies_on_c1", [r.id for r in svc.list_replies(c1.id)])
    print("thread", svc.get_thread("post-1"))

Tests

python

from __future__ import annotations

import threading
import unittest

from service import CommentError, CommentService


class CommentServiceTests(unittest.TestCase):
    def setUp(self):
        self.svc = CommentService()
        self.svc.ensure_post("p1")

    def test_order_independent_of_clock(self):
        ids = []
        for i in range(5):
            ids.append(self.svc.add_comment("p1", "a", f"b{i}").id)
        listed = [c.id for c in self.svc.list_comments("p1")]
        self.assertEqual(listed, ids)
        self.assertEqual(listed, sorted(listed))

    def test_two_levels_only_and_relationship(self):
        c = self.svc.add_comment("p1", "a", "hi")
        r = self.svc.add_reply(c.id, "b", "yo")
        self.assertEqual(r.comment_id, c.id)
        self.assertEqual(r.post_id, "p1")
        with self.assertRaises(CommentError) as ctx:
            self.svc.add_reply(9999, "b", "nope")
        self.assertEqual(ctx.exception.code, "COMMENT_NOT_FOUND")

    def test_validation(self):
        with self.assertRaises(CommentError):
            self.svc.add_comment("p1", "", "x")
        with self.assertRaises(CommentError):
            self.svc.add_comment("p1", "a", " ")
        with self.assertRaises(CommentError):
            self.svc.add_comment("missing", "a", "x")

    def test_independent_id_sequences(self):
        c1 = self.svc.add_comment("p1", "a", "1")
        r1 = self.svc.add_reply(c1.id, "b", "r")
        c2 = self.svc.add_comment("p1", "a", "2")
        # comment ids and reply ids increment independently
        self.assertEqual(c1.id, 1)
        self.assertEqual(c2.id, 2)
        self.assertEqual(r1.id, 1)

    def test_post_isolation(self):
        self.svc.ensure_post("p2")
        self.svc.add_comment("p1", "a", "only-p1")
        self.svc.add_comment("p2", "a", "only-p2")
        self.assertEqual(len(self.svc.list_comments("p1")), 1)
        self.assertEqual(len(self.svc.list_comments("p2")), 1)
        self.assertEqual(self.svc.list_comments("p1")[0].body, "only-p1")

    def test_defensive_snapshot(self):
        self.svc.add_comment("p1", "a", "x")
        thread = self.svc.get_thread("p1")
        thread[0]["body"] = "MUTATED"
        thread[0]["replies"].append({"id": 99})
        again = self.svc.get_thread("p1")
        self.assertEqual(again[0]["body"], "x")
        self.assertEqual(again[0]["replies"], [])

    def test_concurrent_ids_unique(self):
        errors = []
        n_threads = 10
        per = 50

        def worker():
            try:
                for i in range(per):
                    self.svc.add_comment("p1", "a", f"c-{threading.get_ident()}-{i}")
            except Exception as e:
                errors.append(e)

        threads = [threading.Thread(target=worker) for _ in range(n_threads)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        self.assertEqual(errors, [])
        comments = self.svc.list_comments("p1")
        self.assertEqual(len(comments), n_threads * per)
        ids = [c.id for c in comments]
        self.assertEqual(len(ids), len(set(ids)))
        self.assertEqual(ids, sorted(ids))


if __name__ == "__main__":
    unittest.main(verbosity=2)

Test run: Ran 7 tests in 0.005s — OK.

TypeScript sketch

typescript

// Sandbox sketch: mirrors the Python structure (Map + arrays + mutex via async queue or mutexify).
type Comment = { id: number; postId: string; author: string; body: string; createdSeq: number; replies: Reply[] };
type Reply = { id: number; commentId: number; postId: string; author: string; body: string; createdSeq: number };

class CommentService {
  private commentSeq = 0;
  private replySeq = 0;
  private orderSeq = 0;
  private posts = new Map<string, Comment[]>();
  private byId = new Map<number, Comment>();

  ensurePost(postId: string) {
    if (!this.posts.has(postId)) this.posts.set(postId, []);
  }

  addComment(postId: string, author: string, body: string): Omit<Comment, "replies"> {
    const list = this.posts.get(postId);
    if (!list) throw new Error("POST_NOT_FOUND");
    this.commentSeq += 1; this.orderSeq += 1;
    const c: Comment = { id: this.commentSeq, postId, author, body, createdSeq: this.orderSeq, replies: [] };
    list.push(c); this.byId.set(c.id, c);
    return { id: c.id, postId, author, body, createdSeq: c.createdSeq };
  }

  addReply(commentId: number, author: string, body: string): Reply {
    const parent = this.byId.get(commentId);
    if (!parent) throw new Error("COMMENT_NOT_FOUND");
    this.replySeq += 1; this.orderSeq += 1;
    const r: Reply = { id: this.replySeq, commentId, postId: parent.postId, author, body, createdSeq: this.orderSeq };
    parent.replies.push(r);
    return { ...r };
  }

  listComments(postId: string) {
    const list = this.posts.get(postId);
    if (!list) throw new Error("POST_NOT_FOUND");
    return list.map(({ id, postId, author, body, createdSeq }) => ({ id, postId, author, body, createdSeq }));
  }
}

Concepts used, learn more

Read the underlying idea on its own study page. This lesson applies it. It does not replace those pages.

Interview Q&A

Why validate author length before the lock?

Show

Answer

A bad author should not wait on the lock and should not bump a counter. The missing-parent checks stay inside the lock because they read shared maps.

What does a reply store for post_id?

Show

Answer

The parent comment's post_id, copied at insert time, so the reply does not need a second lookup to prove which post it belongs to.

Which pages explain the lock and the interface style?

Show

Answer

What does test_two_levels_only_and_relationship check?

Show

Answer

A reply stores the parent comment id and that comment's post id. add_reply(9999) raises COMMENT_NOT_FOUND. There is no API that attaches a reply to a reply.

Why is get_thread a list of dicts instead of frozen views?

Show

Answer

The nested shape is easier to assert in tests and to sketch in TypeScript. deepcopy detaches the reply arrays. list_comments stays on frozen dataclasses because it is flat.

What would a shared comment and reply counter break?

Show

Answer

test_independent_id_sequences expects the first reply id to be 1 even after a comment id 1 exists. One counter would make that reply id 2.

Pitfalls

Mutating a returned thread dict and corrupting the store (fixed with deepcopy / frozen views).
Sharing one counter for comments and replies when the prompt asks for independent sequences.

The series pager also walks these pages.

Comments and Replies - In-Memory Solution, Ordering & Tests

Question ladder

Failure modes

Misconceptions

Interviewer traps

Design scenario

Frozen views and a deep-copied thread

Return the internal list

Overview

Diagrams

Solution (Python)

Tests

TypeScript sketch

Concepts used, learn more

Interview Q&A

Pitfalls

Related

Comments and Replies - In-Memory Solution, Ordering & Tests

Question ladder

Failure modes

Misconceptions

Interviewer traps

Design scenario

Frozen views and a deep-copied thread

Return the internal list

Overview

Diagrams

Solution (Python)

Tests

TypeScript sketch

Concepts used, learn more

Interview Q&A

Pitfalls

Related