Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion PROJECT_ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,10 @@ Owns:
- Reward minting with admin credentials.
- Trustline submit relay endpoint (for signed XDR).
- Persistent JSON data (`faucet claims`, `artist profile`, `listings`) through `lib/server-data-paths.ts`.
- Signals, replies, edit history and reactions live in SQLite (`lib/sqlite-db.ts`, WAL mode), not the JSON sidecar.
- 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`).
- Concurrency safety for the signal store: version-guarded writes (`WHERE id = ? AND version = ?`) with bounded retry, plus `BEGIN IMMEDIATE` transactions for parent-version checks. Holds across processes, so concurrent Vercel instances no longer interleave. See `docs/TECHNICAL.md` 5.3.
- Other JSON-backed stores (`follow`, `profile`, `market`, `notification`, `achievement`, `narrative-world`) still do unguarded read-modify-write on their sidecar files. Single-writer is safe; concurrent writers lose updates.

Must not own:

Expand Down
26 changes: 7 additions & 19 deletions app/api/faucet/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ async function readClaims(): Promise<FaucetClaims> {
const fp = value.faucetPending
let faucetPending: WalletClaims["faucetPending"]
if (fp && typeof fp === "object" && typeof fp.hash === "string" && typeof fp.at === "number") {
const r = parseRewardType(fp.reward)
const r = await parseRewardType(fp.reward)
faucetPending = { hash: fp.hash, reward: r, at: fp.at }
}
normalized[wallet] = {
Expand Down Expand Up @@ -195,17 +195,6 @@ function serverTokenContractId(): string {
return PHASE_LIQ_TOKEN_CONTRACT
}

function rewardAmountStroops(reward: RewardType): string {
if (reward === "genesis") return PHASER_FAUCET_MINT_STROOPS
if (reward === "daily") return DAILY_REWARD_STROOPS
if (reward === "quest_first_world" || reward === "quest_three_collections") return NEW_QUEST_REWARD_STROOPS
return QUEST_REWARD_STROOPS
}

function isQuestReward(reward: RewardType): reward is QuestId {
return QUEST_IDS.includes(reward as QuestId)
}

async function rewardAmountStroops(reward: RewardType): Promise<string> {
if (reward === "genesis") return PHASER_FAUCET_MINT_STROOPS
if (reward === "daily") return DAILY_REWARD_STROOPS
Expand Down Expand Up @@ -265,7 +254,6 @@ async function buildWalletStatus(wallet: string | null, claims: FaucetClaims) {
claimStatusForReward(claim, "genesis", now),
claimStatusForReward(claim, "daily", now),
])

const rewards: Record<string, RewardStatus> = {
genesis: rawGenesis,
daily: rawDaily,
Expand Down Expand Up @@ -570,7 +558,7 @@ export async function POST(req: NextRequest) {
/** Cada intento de claim debe ver el ledger al día (p. ej. acabas de hacer settle). */
questProgressCache.delete(userAddress)

const reward = parseRewardType(body.reward)
const reward = await parseRewardType(body.reward)
const claims = await readClaims()
const walletClaim = claims[userAddress] ?? {}

Expand Down Expand Up @@ -601,7 +589,7 @@ export async function POST(req: NextRequest) {
}
}

const status = claimStatusForReward(walletClaim, reward, Date.now())
const status = await claimStatusForReward(walletClaim, reward, Date.now())
if (!status.claimable) {
const already = status.nextAt
? "Reward on cooldown. Claim it again after reset."
Expand Down Expand Up @@ -776,7 +764,7 @@ export async function POST(req: NextRequest) {
ok: true,
hash: pendingHash,
reward,
amountStroops: rewardAmountStroops(reward),
amountStroops: await rewardAmountStroops(reward),
...(streakInfo ? { streak: streakInfo } : {}),
...(referralBonus ? { referralBonus } : {}),
})
Expand All @@ -791,7 +779,7 @@ export async function POST(req: NextRequest) {
hash: pendingHash,
pending: true,
reward,
amountStroops: rewardAmountStroops(reward),
amountStroops: await rewardAmountStroops(reward),
note: "Transaction still pending on ledger. Retry the same reward in a few seconds.",
},
{ status: 202 },
Expand Down Expand Up @@ -834,7 +822,7 @@ export async function POST(req: NextRequest) {
}
const c = new Contract(tokenId)
// phase-131: apply streak multiplier to daily reward amount
let effectiveAmountStroops = rewardAmountStroops(reward)
let effectiveAmountStroops = await rewardAmountStroops(reward)
if (reward === "daily" && isStreakMultiplierEnabled()) {
const streak = await getStreakInfo(userAddress)
if (streak.multiplier > 1) {
Expand Down Expand Up @@ -911,7 +899,7 @@ export async function POST(req: NextRequest) {
hash,
pending: true,
reward,
amountStroops: rewardAmountStroops(reward),
amountStroops: await rewardAmountStroops(reward),
note: "Transaction still pending on ledger. Retry the same reward in a few seconds.",
},
{ status: 202 },
Expand Down
3 changes: 2 additions & 1 deletion app/api/profile/avatar/route.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { NextRequest } from "next/server"
import { DEFAULT_PROFILE_LOCALE, getProfile, isProfilePinningRedundancyEnabled, isPhase137Enabled, localizeAvatarName, normalizeProfileLocale, ProfileError, resolveAvatarWithFallback, toProfileErrorResponse } from "@/lib/profile-store"
import { DEFAULT_PROFILE_LOCALE, getProfile, isProfilePinningRedundancyEnabled, isPhase137Enabled, localizeAvatarName, normalizeProfileLocale, ProfileError, resolveAvatarWithFallback, toProfileErrorResponse, BatchAvatarQuerySchema, getAvatarsForWallets } from "@/lib/profile-store"
import { isNftGridVirtualizationEnabled } from "@/lib/nft-grid-virtualization"
import { StrKey } from "@stellar/stellar-sdk"
import { createApiRequestContext } from "@/lib/api-observability"
import { z } from "zod"
Expand Down
2 changes: 0 additions & 2 deletions app/api/profile/follow/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,6 @@ import {
getRequestCost,
FollowSuggestionQuerySchema,
validateSep50MetadataBeforePin,
getFollowSuggestions,
FollowSuggestionQuerySchema,
} from "@/lib/follow-store";
import { createNotification } from "@/lib/notification-store";
import { getProfile } from "@/lib/profile-store";
Expand Down
22 changes: 19 additions & 3 deletions app/api/signals/[id]/replies/route.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,16 @@
import { NextRequest } from "next/server"
import { NextRequest, NextResponse } from "next/server"
import { StrKey } from "@stellar/stellar-sdk"
import { getSignal, createReply, VersionConflictError } from "@/lib/signal-store"
import {
getSignal,
createReply,
VersionConflictError,
isPhase136Enabled,
extractIpfsCidPath,
resolveCidGateway,
recordReplyAttribution,
} from "@/lib/signal-store"
import { computeCreditLedger, getSignalContributors } from "@/lib/contributor-ledger"
import { isFaucetDenyListEnabled, isWalletDenied, getWalletDenyEntry } from "@/lib/faucet-deny-list"
import { createNotification } from "@/lib/notification-store"
import { dispatchPushNotification, extractMentionedWallets, isPhase92Enabled } from "@/lib/push-notifications"
import { createApiRequestContext } from "@/lib/api-observability"
Expand Down Expand Up @@ -33,6 +43,12 @@ type ReplyBody = {
wallet?: unknown
signature?: unknown
parent_version?: unknown
/** Covered by the signed payload; rejected when absent or non-finite. */
timestamp?: unknown
/** phase-116 contributor attribution. Accepts the array directly or wrapped. */
attribution?: unknown
/** Legacy alias for `attribution`; either is honoured. */
contributors?: unknown
}

const ContributorsArraySchema = z.array(
Expand Down Expand Up @@ -200,7 +216,7 @@ export async function POST(
{
error: "Signal changed since this reply was composed",
code: "VERSION_CONFLICT",
currentVersion: signal.version,
current_version: signal.version,
},
{ status: 409, event: "signals.reply.version_conflict" },
)
Expand Down
4 changes: 3 additions & 1 deletion app/api/signals/[id]/route.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { NextRequest } from "next/server"
import { NextRequest, NextResponse } from "next/server"
import { StrKey } from "@stellar/stellar-sdk"
import {
getSignal,
Expand All @@ -7,6 +7,8 @@ import {
signalETag,
parseVersionHeader,
VersionConflictError,
editSignal,
SignalEditError,
} from "@/lib/signal-store"
import { createNotification } from "@/lib/notification-store"
import { checkAndUnlock } from "@/lib/achievement-store"
Expand Down
6 changes: 3 additions & 3 deletions app/api/soroban-rpc/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ const CIRCUIT_COOLDOWN_MS = 10_000
let inFlight = 0
let consecutiveFailures = 0
let circuitOpenedAt = 0

/** Debe cubrir varias URLs × reintentos × timeout (p. ej. 3×2×45s); Vercel Pro permite hasta 300s. */
export const maxDuration = 300

Expand Down Expand Up @@ -101,7 +100,8 @@ interface CircuitStatus {
}

const upstreamCircuits = new Map<string, CircuitStatus>()
const CIRCUIT_COOLDOWN_MS = 15_000
/** Per-upstream cooldown, distinct from the process-wide CIRCUIT_COOLDOWN_MS above. */
const UPSTREAM_CIRCUIT_COOLDOWN_MS = 15_000
const FAILURE_THRESHOLD = 3

export function getUpstreamCircuitStatus(url: string): CircuitStatus {
Expand All @@ -127,7 +127,7 @@ export function recordUpstreamFailure(url: string, is429: boolean = false) {
status.failures += 1
if (is429 || status.failures >= FAILURE_THRESHOLD) {
status.state = "OPEN"
status.nextAttempt = Date.now() + CIRCUIT_COOLDOWN_MS
status.nextAttempt = Date.now() + UPSTREAM_CIRCUIT_COOLDOWN_MS
}
}

Expand Down
1 change: 1 addition & 0 deletions app/signals/[id]/page.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import { notFound } from "next/navigation"
import Link from "next/link"
import { WalletAvatar } from "@/components/wallet-avatar"
import { getSignal, getReplies, getSignalContributors, computeCreditLedger, isPhase116Enabled, isPhase136Enabled, resolveCidGateway, extractIpfsCidPath } from "@/lib/signal-store"
import { isFaucetDenyListEnabled, isWalletDenied } from "@/lib/faucet-deny-list"
import { SignalDetailClient } from "./signal-detail-client"

export const dynamic = "force-dynamic"
Expand Down
2 changes: 2 additions & 0 deletions app/signals/[id]/signal-detail-client.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ const copy = {
ctaBusy: "[ SENDING… ]",
noWallet: "[ CONNECT_WALLET_TO_REPLY ]",
walletBadge: "✓ WALLET",
verifiedBadge: "✓ VERIFIED",
conflict: "[ SIGNAL_CHANGED_REFRESH_AND_RETRY ]",
},
es: {
Expand All @@ -26,6 +27,7 @@ const copy = {
ctaBusy: "[ ENVIANDO… ]",
noWallet: "[ CONECTAR_WALLET_PARA_RESPONDER ]",
walletBadge: "✓ WALLET",
verifiedBadge: "✓ VERIFICADO",
conflict: "[ SEÑAL_ACTUALIZADA_REFRESCA_Y_REINTENTA ]",
},
}
Expand Down
58 changes: 36 additions & 22 deletions docs/TECHNICAL.md
Original file line number Diff line number Diff line change
Expand Up @@ -180,39 +180,53 @@ settlement verifier and must not accept unsigned base64 payloads.

### 5.3 Signal concurrency contract

`lib/signal-store.ts` guards the JSON sidecar three ways:
Signals and replies are stored in SQLite (`signals`, `signal_replies`,
`signal_versions`, `signal_reactions` in `lib/sqlite-db.ts`), not in a JSON
sidecar. The connection runs in WAL mode with `foreign_keys = ON`.

1. **Serialized mutation.** Read-modify-write cycles run through a per-file
promise queue, so two concurrent mutations cannot both read the same snapshot.
2. **Atomic replacement.** Writes land in a sibling `*.tmp` file and are moved
into place with `rename(2)`. A reader never observes a half-flushed file, and
a crash mid-write cannot corrupt the live store.
3. **Corruption is fatal, not silent.** A JSON parse failure throws instead of
degrading to `{}`. Returning `{}` on a parse error meant the next write would
overwrite every existing record.
Each `Signal` carries an integer `version` in the `signals` table, starting at
`1` and bumped on every mutation. `1` matters: the version is published as an
ETag and echoed back as `If-Match` / `parent_version`, and all three reject `0`.

Each `Signal` carries an integer `version`, bumped on every mutation.
The lost-update hazard is `upvoteSignal`, the only read-modify-write over a
JSON blob (`upvotes_json`). It is guarded three ways:

1. **Version-guarded write.** The `UPDATE` names the version it read:
`... WHERE id = ? AND version = ?`. A writer whose row was superseded
matches zero rows instead of clobbering.
2. **Bounded retry.** `upvoteSignal` re-reads and recomputes on a lost CAS,
up to `MAX_CAS_ATTEMPTS` (8). Because the guard is version-based rather
than wall-clock, retries converge.
3. **Transactional parent check.** `createReply` and `editSignal` verify the
parent version inside the same `BEGIN IMMEDIATE` as their write, so the
check cannot be invalidated between the test and the write.

A stale value throws `VersionConflictError`, which the API maps to `409`.

| Surface | Contract |
|---|---|
| `GET /api/signals/[id]` | Returns `ETag: "<version>"`. |
| `POST /api/signals/[id]` | Optional `If-Match: "<version>"`. Stale value → `409` with `current_version` and the fresh `ETag`. |
| `POST /api/signals/[id]/replies` | Optional `parent_version: <int>`. Stale value → `409`. |
| `POST /api/signals/[id]/replies` | Optional `parent_version: <int>`. Stale value → `409` with `current_version`. |
| `PATCH /api/signals/[id]` | `If-Match` required (phase-82). Stale value → `409`. |

`If-Match` and `parent_version` are both optional, so clients that omit them keep
the previous last-writer-wins behaviour. Conflicts emit the
`If-Match` and `parent_version` are optional on the upvote and reply paths, so
clients that omit them keep last-writer-wins behaviour. Conflicts emit the
`signals.version_conflict` log event.

Records written before this change have no `version`; they are normalized to `1`
on read, so no migration step is required.
Rows written before the column existed are normalized to `1` by an idempotent
migration in `getDb()`.

**Cross-instance behaviour.** Unlike a per-process lock, the version guard and
WAL transactions hold across processes, so two Vercel instances sharing the
database file no longer interleave destructively. `lib/__tests__/signal-store-concurrency.test.ts`
drives this directly with a second `DatabaseSync` handle committing between one
connection's read and its write.

**Known limitation.** The mutation queue is per-process. Two Vercel instances
mutating the same file can still interleave, because each holds its own queue and
the deployment shares no writable volume by default. Closing that requires either
pinning writes to a single instance or moving signals to a store with real
cross-process transactions (Postgres `UPDATE ... WHERE version = $expected`).
This is deliberate: it is a bounded, documented limit rather than a CRDT layer
that would not have fixed the underlying file race.
**Not a CRDT.** Signals are versioned, not collaboratively edited: an author's
edit takes a pre-edit snapshot into `signal_versions` and must stay
revertible. A merge-based CRDT would fold concurrent edits together and destroy
that history, so the contract is compare-and-swap rather than automatic merge.

---

Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/artist-badge-attestation.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ import assert from "node:assert/strict"
import { mkdtemp, rm } from "node:fs/promises"
import os from "node:os"
import path from "node:path"
import { after, before, beforeEach, describe, it } from "node:test"
import { afterAll as after, beforeAll as before, beforeEach, describe, it } from "vitest"
import { Keypair } from "@stellar/stellar-sdk"
import {
ArtistAttestationError,
Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/avatar-provenance.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
* Run with: npm test
*/
import assert from "node:assert/strict"
import { describe, it } from "node:test"
import { describe, it } from "vitest"
import { Keypair } from "@stellar/stellar-sdk"
import { CidIntegrityError, sha256Hex, verifyCID, verifyBytesIntegrity } from "@/lib/cid-cache"

Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/cid-gateway-cache.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it, beforeEach } from "node:test"
import { describe, it, beforeEach } from "vitest"
import * as assert from "node:assert/strict"
import {
resolveCidGateway,
Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/classic-liq-batch.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it } from "node:test"
import { describe, it } from "vitest"
import * as assert from "node:assert/strict"
import { submitTrustlineBatch } from "@/lib/classic-liq"

Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/collection-offer-book.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ import assert from "node:assert/strict"
import { mkdtemp, rm } from "node:fs/promises"
import os from "node:os"
import path from "node:path"
import { after, before, describe, it } from "node:test"
import { afterAll as after, beforeAll as before, describe, it } from "vitest"
import { Keypair } from "@stellar/stellar-sdk"
import {
createListing,
Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/cost-attribution-ledger.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it, beforeEach } from "node:test"
import { describe, it, beforeEach } from "vitest"
import * as assert from "node:assert/strict"
import {
recordRequestCost,
Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/custodian-release-authorization.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it } from "node:test"
import { describe, it } from "vitest"
import * as assert from "node:assert/strict"
import { isCustodianReleaseAuthorized } from "@/lib/phase-protocol"

Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/explore-dedup.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it, beforeEach } from "node:test"
import { describe, it, beforeEach } from "vitest"
import * as assert from "node:assert/strict"
import { assetContentHash, dedupeExploreItems, type ExploreItem } from "@/lib/explore-domain"

Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/explore-owners-cache.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it, beforeEach } from "node:test"
import { describe, it, beforeEach } from "vitest"
import * as assert from "node:assert/strict"
import {
getCachedExploreOwners,
Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/follow-graph-indexer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
* Validates performance of follow relationship queries and graph traversal
*/

import { describe, it, expect } from "@jest/globals"
import { describe, it, expect } from "vitest"

describe("Follow Graph Indexer Performance (#46)", () => {
it("should calculate follower counts without parsing entire file", async () => {
Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/follow-sep50.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it } from "node:test"
import { describe, it } from "vitest"
import * as assert from "node:assert/strict"
import { validateSep50MetadataBeforePin } from "@/lib/follow-store"

Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/gateway-health.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it, beforeEach } from "node:test"
import { describe, it, beforeEach } from "vitest"
import * as assert from "node:assert/strict"
import { recordGatewayLatency, getGatewayHealthSnapshot, resetGatewayHealth } from "@/lib/gateway-health"

Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/horizon-bulkhead.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { describe, it, beforeEach } from "node:test"
import { describe, it, beforeEach } from "vitest"
import * as assert from "node:assert/strict"
import {
getHorizonBulkheadMetrics,
Expand Down
2 changes: 1 addition & 1 deletion lib/__tests__/i18n-loader.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
* Validates translation loading, caching, and fallback mechanisms
*/

import { describe, it, expect, beforeEach } from "@jest/globals"
import { describe, it, expect, beforeEach } from "vitest"

// Mock translations for testing
const mockTranslations = {
Expand Down
Loading