engineering/remarkbox
1
0
Fork 0
Code Issues 3 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Use embed-identical root node creation when thread_uri is provided

Browse source 
When thread_uri is given to POST /api/v1/threads, the API now uses
get_or_create_node_by_uri (same as the embed iframe) to create the
root node, then posts the comment as a child reply. This ensures
threads created via the API are structurally identical to those
created by the embed, so the iframe can find and display them.

Without thread_uri, behavior is unchanged (standalone root node).
This commit is contained in:
timehexon 2026-02-02 18:47:38 +00:00
parent a3e6c0b046
commit 5042a7d9f1
2 changed files with 74 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

68
remarkbox/api/views.py
View file

@ -23,7 +23,7 @@ from remarkbox.models.namespace import (
get_topsecret_namespaces,
)
from remarkbox.models.uri import get_or_create_uri
from remarkbox.models.node import Node
from remarkbox.models.node import Node, get_or_create_node_by_uri
from remarkbox.lib.mail import send_verification_digits_to_email
from remarkbox.lib.notify import schedule_notifications
from remarkbox.views import verify_pending_nodes_in_session
@ -503,7 +503,55 @@ def api_create_thread(request):
request.response.status_code = 400
return {"error": "email is required (or namespace must allow_anonymous)"}
# Create root node
if thread_uri:
# Simulate what the embed iframe does: get or create the root node
# via URI, then post the comment as a reply under it.
root = get_or_create_node_by_uri(
request.dbsession, thread_uri, node_title=title
)
request.dbsession.add(root)
request.dbsession.flush()
# Create the comment as a child of the root (same as api_reply)
node = root.new_child()
node.ip_address = str(request.client_addr)
node.set_data(data, namespace=namespace, dbsession=request.dbsession)
if user_surrogate:
node.user_surrogate = user_surrogate
node.verified = True
node_event = None
request.dbsession.add(user_surrogate)
else:
node.user = user
node.verified = user.authenticated
node_event = node.new_event(user, "commented")
request.dbsession.add(user)
if spam_held:
node.approved = False
# Bump thread
root.changed = node.changed
root._invalidate_cache()
request.dbsession.add(node)
if node_event:
request.dbsession.add(node_event)
request.dbsession.add(root)
request.dbsession.flush()
if node_event:
schedule_notifications(request, node_event)
request.response.status_code = 201
return {
"node": serialize_node(node),
"root_node_id": str(root.id),
"verified": node.verified,
}
# No thread_uri — create a standalone thread (root node holds the content)
node = create_root_node()
node.namespace = namespace
node.ip_address = str(request.client_addr)
@ -530,22 +578,6 @@ def api_create_thread(request):
request.dbsession.add(namespace)
request.dbsession.flush()
# Link thread to a page URI so the embed iframe can find it
if thread_uri:
uri = get_or_create_uri(request.dbsession, thread_uri)
if uri.node is not None and uri.node.id != node.id:
# URI already has a thread — caller should use the reply endpoint
request.response.status_code = 409
return {
"error": "A thread already exists for this URI",
"existing_node_id": str(uri.node.id),
}
uri.node = node
node.has_uri = True
request.dbsession.add(uri)
request.dbsession.add(node)
request.dbsession.flush()
if node_event:
request.node = node
schedule_notifications(request, node_event)

35
remarkbox/tests/test_api_views.py
View file

@ -98,6 +98,8 @@ class TestAPIAnonymousPosting(APIFunctionalTests):
self.assertEqual(body["node"]["author"]["name"], "ClaudeBot")
def test_create_thread_with_uri(self):
"""thread_uri creates a root via get_or_create_node_by_uri and
returns the comment as a child reply — same as the embed iframe."""
res = self.testapp.post_json(
"/api/v1/threads",
{
@ -110,43 +112,54 @@ class TestAPIAnonymousPosting(APIFunctionalTests):
expect_errors=True,
)
self.assertEqual(res.status_int, 201)
node_id = res.json["node"]["id"]
reply_id = res.json["node"]["id"]
root_id = res.json["root_node_id"]
# Verify the URI record was created and linked
# The reply should be a child, not the root itself
self.assertNotEqual(reply_id, root_id)
# Verify the URI record links to the root node (not the reply)
from remarkbox.models.uri import get_uri_by_uri
uri = get_uri_by_uri(self.dbsession, "https://api-test.example.com/my-page/")
self.assertIsNotNone(uri)
self.assertEqual(str(uri.node.id), node_id)
self.assertEqual(str(uri.node.id), root_id)
def test_create_thread_duplicate_uri(self):
# First thread with URI
def test_create_thread_duplicate_uri_adds_reply(self):
"""Posting to the same thread_uri twice adds a second reply
under the same root — it does not fail with 409."""
res1 = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "First",
"data": "First thread",
"data": "First comment",
"thread_uri": "https://api-test.example.com/dup-test/",
"anonymous_name": "Bot",
},
expect_errors=True,
)
self.assertEqual(res1.status_int, 201)
root_id_1 = res1.json["root_node_id"]
# Second thread with same URI should fail
# Second post to same URI creates another reply under the same root
res2 = self.testapp.post_json(
"/api/v1/threads",
{
"namespace": self.namespace_name,
"title": "Second",
"data": "Duplicate",
"title": "First",
"data": "Second comment",
"thread_uri": "https://api-test.example.com/dup-test/",
"anonymous_name": "Bot",
},
expect_errors=True,
)
self.assertEqual(res2.status_int, 409)
self.assertIn("already exists", res2.json["error"])
self.assertEqual(res2.status_int, 201)
root_id_2 = res2.json["root_node_id"]
# Both replies share the same root
self.assertEqual(root_id_1, root_id_2)
# But the reply nodes are different
self.assertNotEqual(res1.json["node"]["id"], res2.json["node"]["id"])
def test_create_anonymous_thread_default_name(self):
res = self.testapp.post_json(