fix(signals): stop losing writes on concurrent signal mutations

PROJECT_ARCHITECTURE.md

@@ -52,6 +52,7 @@ Owns:
 - Trustline submit relay endpoint (for signed XDR).
 - Persistent JSON data (`faucet claims`, `artist profile`, `listings`) through `lib/server-data-paths.ts`.
 - x402 endpoints and payment verification support.
+- Concurrency safety for JSON-backed stores: serialized read-modify-write plus atomic file replacement (see `lib/signal-store.ts`).
 
 Must not own:
 
@@ -95,6 +96,12 @@ Owns:
 - **Deterministic status codes** (validation, cooldown, authorization, pending).
 - **No implicit success**: all state transitions explicit and auditable.
 - **Compatibility routes** allowed if typed and documented.
+- **Optimistic concurrency on shared mutable records**: a record that more than one
+  client can edit carries an integer `version`. Reads publish it as an `ETag`;
+  mutations may echo it back as `If-Match` (or `parent_version` for appends) and
+  the server rejects a stale value with `409` rather than silently overwriting a
+  concurrent writer. `409` is logged under the `signals.version_conflict` event
+  so conflict rates are observable.
 
 ## 6) Internationalization architecture
 

app/api/signals/[id]/replies/route.ts

@@ -1,6 +1,6 @@
 import { NextRequest } from "next/server"
 import { StrKey } from "@stellar/stellar-sdk"
-import { getSignal, createReply, AttributionInReplySchema, recordReplyAttribution, getSignalContributors, computeCreditLedger, isPhase136Enabled, resolveCidGateway, extractIpfsCidPath } from "@/lib/signal-store"
+import { getSignal, createReply, VersionConflictError } from "@/lib/signal-store"
 import { createNotification } from "@/lib/notification-store"
 import { dispatchPushNotification, extractMentionedWallets, isPhase92Enabled } from "@/lib/push-notifications"
 import { createApiRequestContext } from "@/lib/api-observability"
@@ -32,10 +32,6 @@ type ReplyBody = {
   body?: unknown
   wallet?: unknown
   signature?: unknown
-  timestamp?: unknown
-  attribution?: unknown
-  contributors?: unknown
-  /** Issue #224: signal version this reply was composed against (CAS). */
   parent_version?: unknown
 }
 
@@ -161,6 +157,21 @@ export async function POST(
     }
   }
 
+  // Optional: when present, the reply must be composed against this version of
+  // the parent signal. A stale parent_version is rejected rather than silently
+  // accepted against a view the author never saw.
+  let parentVersion: number | undefined
+  if (body.parent_version !== undefined) {
+    const parsed = body.parent_version
+    if (typeof parsed !== "number" || !Number.isInteger(parsed) || parsed < 1) {
+      return api.json(
+        { error: "Invalid parent_version" },
+        { status: 400, event: "signals.reply.validation_failed", metadata: { reason: "parent_version" } },
+      )
+    }
+    parentVersion = parsed
+  }
+
   try {
     const signal = await getSignal(id)
     if (!signal) {
@@ -211,15 +222,17 @@ export async function POST(
       }
     }
 
-    const reply = await createReply({
-      signal_id: id,
-      author_wallet: walletStr,
-      author_display,
-      body: (body.body as string).trim(),
-      upvotes: [],
-      signature: body.signature as string,
-      signature_verified: signatureVerified,
-    })
+    const reply = await createReply(
+      {
+        signal_id: id,
+        author_wallet: walletStr,
+        author_display,
+        body: (body.body as string).trim(),
+        upvotes: [],
+        signature: body.signature as string,
+      },
+      parentVersion,
+    )
 
     // phase-116: record contributor attribution (flag-gated, best-effort)
     let creditLedger: Awaited<ReturnType<typeof computeCreditLedger>> | null = null
@@ -300,37 +313,16 @@ export async function POST(
       },
     )
   } catch (error) {
+    if (error instanceof VersionConflictError) {
+      return api.json(
+        { error: "Version conflict", current_version: error.currentVersion },
+        {
+          status: 409,
+          event: "signals.version_conflict",
+          metadata: { signal_id: id, route: "replies", current_version: error.currentVersion },
+        },
+      )
+    }
     return api.errorJson(error, 500, "signals.reply.create_failed")
   }
 }
-
-export async function GET(
-  request: NextRequest,
-  { params }: { params: Promise<{ id: string }> },
-) {
-  const api = createApiRequestContext(request, "/api/signals/[id]/replies")
-  const { id } = await params
-  if (!isPhase116Enabled()) {
-    return api.json({ error: "Contributor ledger disabled (phase-116 flag off)" }, { status: 404, event: "signals.ledger.disabled" })
-  }
-  try {
-    const { getSignal } = await import("@/lib/signal-store")
-    const signal = await getSignal(id)
-    if (!signal) return api.json({ error: "Signal not found" }, { status: 404, event: "signals.ledger.signal_missing" })
-    const contributors = await getSignalContributors(id)
-    const creditLedger = await computeCreditLedger(id)
-    const resolvedImage = resolveSignalImage(signal.nft_image)
-    return api.json(
-      {
-        signalId: id,
-        contributors: contributors?.contributors ?? [],
-        totalShareBps: contributors?.totalShareBps ?? 0,
-        creditLedger,
-        ...(resolvedImage ? { signalMedia: resolvedImage } : {}),
-      },
-      { event: "signals.ledger.loaded", metadata: { signal_id: id, phase136: isPhase136Enabled() } },
-    )
-  } catch (error) {
-    return api.errorJson(error, 500, "signals.ledger.load_failed")
-  }
-}

app/api/signals/[id]/route.ts

@@ -1,23 +1,38 @@
-import { NextRequest, NextResponse } from "next/server"
+import { NextRequest } from "next/server"
 import { StrKey } from "@stellar/stellar-sdk"
-import { getSignal, upvoteSignal, getReplies, editSignal, SignalEditError } from "@/lib/signal-store"
+import {
+  getSignal,
+  upvoteSignal,
+  getReplies,
+  signalETag,
+  parseVersionHeader,
+  VersionConflictError,
+} from "@/lib/signal-store"
 import { createNotification } from "@/lib/notification-store"
 import { checkAndUnlock } from "@/lib/achievement-store"
+import { createApiRequestContext } from "@/lib/api-observability"
 
 export const runtime = "nodejs"
 export const dynamic = "force-dynamic"
 
 export async function GET(
-  _request: NextRequest,
+  request: NextRequest,
   { params }: { params: Promise<{ id: string }> },
 ) {
+  const api = createApiRequestContext(request, "/api/signals/[id]")
   const { id } = await params
   const signal = await getSignal(id)
   if (!signal) {
-    return NextResponse.json({ error: "Signal not found" }, { status: 404 })
+    return api.json(
+      { error: "Signal not found" },
+      { status: 404, event: "signals.get.not_found", metadata: { signal_id: id } },
+    )
   }
   const replies = await getReplies(id)
-  return NextResponse.json({ signal, replies })
+  return api.json(
+    { signal, replies },
+    { status: 200, event: "signals.get.ok", headers: { ETag: signalETag(signal.version) } },
+  )
 }
 
 type UpvoteBody = {
@@ -29,39 +44,88 @@ export async function POST(
   request: NextRequest,
   { params }: { params: Promise<{ id: string }> },
 ) {
+  const api = createApiRequestContext(request, "/api/signals/[id]")
   const { id } = await params
   let body: UpvoteBody
   try {
     body = (await request.json()) as UpvoteBody
   } catch {
-    return NextResponse.json({ error: "Invalid JSON" }, { status: 400 })
+    return api.json(
+      { error: "Invalid JSON" },
+      { status: 400, event: "signals.upvote.invalid_json" },
+    )
   }
 
   if (typeof body.wallet !== "string" || !StrKey.isValidEd25519PublicKey(body.wallet)) {
-    return NextResponse.json({ error: "Invalid wallet address" }, { status: 400 })
+    return api.json(
+      { error: "Invalid wallet address" },
+      { status: 400, event: "signals.upvote.validation_failed", metadata: { reason: "wallet" } },
+    )
   }
   if (typeof body.signature !== "string" || body.signature.length === 0) {
-    return NextResponse.json({ error: "Signature required" }, { status: 400 })
+    return api.json(
+      { error: "Signature required" },
+      { status: 400, event: "signals.upvote.validation_failed", metadata: { reason: "signature" } },
+    )
+  }
+
+  // If-Match is optional so existing clients keep working; when supplied the
+  // upvote becomes a compare-and-swap against the version the client rendered.
+  const ifMatch = request.headers.get("if-match")
+  let expectedVersion: number | undefined
+  if (ifMatch !== null && ifMatch.trim() !== "*") {
+    const parsed = parseVersionHeader(ifMatch)
+    if (parsed === null) {
+      return api.json(
+        { error: "Invalid If-Match" },
+        { status: 400, event: "signals.upvote.validation_failed", metadata: { reason: "if_match" } },
+      )
+    }
+    expectedVersion = parsed
   }
 
   try {
-    const signal = await upvoteSignal(id, body.wallet)
+    const signal = await upvoteSignal(id, body.wallet, expectedVersion)
     // Notify at milestones: 5, 10, 25 upvotes (fire-and-forget)
     const count = signal.upvotes.length
     if ((count === 5 || count === 10 || count === 25) && signal.author_wallet !== body.wallet) {
       void createNotification(signal.author_wallet, "signal_upvote", {
         signal_id: id,
         signal_title: signal.title,
         upvote_count: count,
-      }).catch(() => { /* silent */ })
+      }).catch((error) => api.log("warn", "signals.upvote.notification_failed", { error }))
     }
     // Achievement: track upvotes for the author (fire-and-forget)
     if (signal.author_wallet !== body.wallet) {
-      void checkAndUnlock(signal.author_wallet, { upvote_delta: 1 }).catch(() => { /* silent */ })
+      void checkAndUnlock(signal.author_wallet, { upvote_delta: 1 }).catch((error) =>
+        api.log("warn", "signals.upvote.achievement_failed", { error }),
+      )
     }
-    return NextResponse.json({ signal })
-  } catch {
-    return NextResponse.json({ error: "Signal not found" }, { status: 404 })
+    return api.json(
+      { signal },
+      {
+        status: 200,
+        event: "signals.upvote.ok",
+        metadata: { signal_id: id, version: signal.version },
+        headers: { ETag: signalETag(signal.version) },
+      },
+    )
+  } catch (error) {
+    if (error instanceof VersionConflictError) {
+      return api.json(
+        { error: "Version conflict", current_version: error.currentVersion },
+        {
+          status: 409,
+          event: "signals.version_conflict",
+          metadata: { signal_id: id, current_version: error.currentVersion },
+          headers: { ETag: signalETag(error.currentVersion) },
+        },
+      )
+    }
+    return api.json(
+      { error: "Signal not found" },
+      { status: 404, event: "signals.upvote.not_found", metadata: { signal_id: id } },
+    )
   }
 }
 

app/signals/[id]/page.tsx

@@ -195,7 +195,11 @@ export default async function SignalDetailPage({ params }: Props) {
         </article>
 
         {/* Replies + compose (client island) */}
-        <SignalDetailClient signalId={id} initialReplies={replies} />
+        <SignalDetailClient
+          signalId={id}
+          initialSignalVersion={signal.version}
+          initialReplies={replies}
+        />
       </div>
     </div>
   )

app/signals/[id]/signal-detail-client.tsx

@@ -16,8 +16,7 @@ const copy = {
     ctaBusy: "[ SENDING… ]",
     noWallet: "[ CONNECT_WALLET_TO_REPLY ]",
     walletBadge: "✓ WALLET",
-    verifiedBadge: "✓ VERIFIED",
-    legacyBadge: "LEGACY",
+    conflict: "[ SIGNAL_CHANGED_REFRESH_AND_RETRY ]",
   },
   es: {
     replies: "RESPUESTAS",
@@ -27,8 +26,7 @@ const copy = {
     ctaBusy: "[ ENVIANDO… ]",
     noWallet: "[ CONECTAR_WALLET_PARA_RESPONDER ]",
     walletBadge: "✓ WALLET",
-    verifiedBadge: "✓ VERIFICADO",
-    legacyBadge: "LEGADO",
+    conflict: "[ SEÑAL_ACTUALIZADA_REFRESCA_Y_REINTENTA ]",
   },
 }
 
@@ -44,25 +42,39 @@ function timeAgo(ts: number): string {
 
 type Props = {
   signalId: string
+  initialSignalVersion: number
   initialReplies: SignalReply[]
 }
 
-export function SignalDetailClient({ signalId, initialReplies }: Props) {
+type PostReplyResponse = {
+  reply?: SignalReply
+  error?: string
+  current_version?: number
+}
+
+export function SignalDetailClient({
+  signalId,
+  initialSignalVersion,
+  initialReplies,
+}: Props) {
   const { address } = useWallet()
   const { lang } = useLang()
   const t = copy[lang] ?? copy.en
 
   const [replies, setReplies] = useState<SignalReply[]>(initialReplies)
+  const [signalVersion, setSignalVersion] = useState(initialSignalVersion)
   const [replyBody, setReplyBody] = useState("")
   const [busy, setBusy] = useState(false)
   const [error, setError] = useState<string | null>(null)
+  const [conflict, setConflict] = useState(false)
 
   const baseInput =
     "w-full bg-transparent border border-[var(--color-border-tertiary)] font-mono text-[12px] text-foreground px-3 py-2 focus:outline-none focus:border-[#7F77DD] transition-colors placeholder:text-muted-foreground/40 resize-none"
 
   async function handleReply() {
     if (!address || !replyBody.trim()) return
     setError(null)
+    setConflict(false)
     setBusy(true)
     try {
       const timestamp = Date.now()
@@ -77,11 +89,19 @@ export function SignalDetailClient({ signalId, initialReplies }: Props) {
         body: JSON.stringify({
           body: replyBodyTrimmed,
           wallet: address,
-          signature,
-          timestamp,
+          // TODO: replace provisional signature with Freighter signMessage when available
+          signature: address,
+          parent_version: signalVersion,
         }),
       })
-      const data = (await res.json().catch(() => ({}))) as { reply?: SignalReply; error?: string }
+      const data = (await res.json().catch(() => ({}))) as PostReplyResponse
+      if (res.status === 409) {
+        if (typeof data.current_version === "number") {
+          setSignalVersion(data.current_version)
+        }
+        setConflict(true)
+        return
+      }
       if (!res.ok || !data.reply) {
         setError(data.error ?? "Reply failed")
         return
@@ -163,6 +183,9 @@ export function SignalDetailClient({ signalId, initialReplies }: Props) {
             {error && (
               <p className="font-mono text-[10px] text-destructive">{error}</p>
             )}
+            {conflict && (
+              <p className="font-mono text-[10px] text-[#7F77DD]">{t.conflict}</p>
+            )}
             <button
               type="button"
               disabled={busy || replyBody.trim().length === 0}