diff --git a/PROJECT_ARCHITECTURE.md b/PROJECT_ARCHITECTURE.md index ad90676..7e17a54 100644 --- a/PROJECT_ARCHITECTURE.md +++ b/PROJECT_ARCHITECTURE.md @@ -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: diff --git a/app/api/faucet/route.ts b/app/api/faucet/route.ts index 38e01da..0afbd89 100644 --- a/app/api/faucet/route.ts +++ b/app/api/faucet/route.ts @@ -140,7 +140,7 @@ async function readClaims(): Promise { 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] = { @@ -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 { if (reward === "genesis") return PHASER_FAUCET_MINT_STROOPS if (reward === "daily") return DAILY_REWARD_STROOPS @@ -265,7 +254,6 @@ async function buildWalletStatus(wallet: string | null, claims: FaucetClaims) { claimStatusForReward(claim, "genesis", now), claimStatusForReward(claim, "daily", now), ]) - const rewards: Record = { genesis: rawGenesis, daily: rawDaily, @@ -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] ?? {} @@ -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." @@ -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 } : {}), }) @@ -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 }, @@ -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) { @@ -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 }, diff --git a/app/api/profile/avatar/route.ts b/app/api/profile/avatar/route.ts index 50b49af..6028b5c 100644 --- a/app/api/profile/avatar/route.ts +++ b/app/api/profile/avatar/route.ts @@ -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" diff --git a/app/api/profile/follow/route.ts b/app/api/profile/follow/route.ts index 9fe8b2c..5c55921 100644 --- a/app/api/profile/follow/route.ts +++ b/app/api/profile/follow/route.ts @@ -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"; diff --git a/app/api/signals/[id]/replies/route.ts b/app/api/signals/[id]/replies/route.ts index be4337a..60c6bf6 100644 --- a/app/api/signals/[id]/replies/route.ts +++ b/app/api/signals/[id]/replies/route.ts @@ -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" @@ -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( @@ -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" }, ) diff --git a/app/api/signals/[id]/route.ts b/app/api/signals/[id]/route.ts index 7d523b9..65268ef 100644 --- a/app/api/signals/[id]/route.ts +++ b/app/api/signals/[id]/route.ts @@ -1,4 +1,4 @@ -import { NextRequest } from "next/server" +import { NextRequest, NextResponse } from "next/server" import { StrKey } from "@stellar/stellar-sdk" import { getSignal, @@ -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" diff --git a/app/api/soroban-rpc/route.ts b/app/api/soroban-rpc/route.ts index 5b2b107..f50038e 100644 --- a/app/api/soroban-rpc/route.ts +++ b/app/api/soroban-rpc/route.ts @@ -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 @@ -101,7 +100,8 @@ interface CircuitStatus { } const upstreamCircuits = new Map() -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 { @@ -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 } } diff --git a/app/signals/[id]/page.tsx b/app/signals/[id]/page.tsx index f0f24ab..b9fbb2c 100644 --- a/app/signals/[id]/page.tsx +++ b/app/signals/[id]/page.tsx @@ -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" diff --git a/app/signals/[id]/signal-detail-client.tsx b/app/signals/[id]/signal-detail-client.tsx index 2b9ff90..10beb22 100644 --- a/app/signals/[id]/signal-detail-client.tsx +++ b/app/signals/[id]/signal-detail-client.tsx @@ -16,6 +16,7 @@ const copy = { ctaBusy: "[ SENDING… ]", noWallet: "[ CONNECT_WALLET_TO_REPLY ]", walletBadge: "✓ WALLET", + verifiedBadge: "✓ VERIFIED", conflict: "[ SIGNAL_CHANGED_REFRESH_AND_RETRY ]", }, es: { @@ -26,6 +27,7 @@ const copy = { ctaBusy: "[ ENVIANDO… ]", noWallet: "[ CONECTAR_WALLET_PARA_RESPONDER ]", walletBadge: "✓ WALLET", + verifiedBadge: "✓ VERIFICADO", conflict: "[ SEÑAL_ACTUALIZADA_REFRESCA_Y_REINTENTA ]", }, } diff --git a/docs/TECHNICAL.md b/docs/TECHNICAL.md index aa025ff..7876fbf 100644 --- a/docs/TECHNICAL.md +++ b/docs/TECHNICAL.md @@ -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: ""`. | | `POST /api/signals/[id]` | Optional `If-Match: ""`. Stale value → `409` with `current_version` and the fresh `ETag`. | -| `POST /api/signals/[id]/replies` | Optional `parent_version: `. Stale value → `409`. | +| `POST /api/signals/[id]/replies` | Optional `parent_version: `. 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. --- diff --git a/lib/__tests__/artist-badge-attestation.test.ts b/lib/__tests__/artist-badge-attestation.test.ts index 7b57a32..880ba60 100644 --- a/lib/__tests__/artist-badge-attestation.test.ts +++ b/lib/__tests__/artist-badge-attestation.test.ts @@ -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, diff --git a/lib/__tests__/avatar-provenance.test.ts b/lib/__tests__/avatar-provenance.test.ts index 1cf9986..c31fa89 100644 --- a/lib/__tests__/avatar-provenance.test.ts +++ b/lib/__tests__/avatar-provenance.test.ts @@ -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" diff --git a/lib/__tests__/cid-gateway-cache.test.ts b/lib/__tests__/cid-gateway-cache.test.ts index 77b4a94..518fa2c 100644 --- a/lib/__tests__/cid-gateway-cache.test.ts +++ b/lib/__tests__/cid-gateway-cache.test.ts @@ -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, diff --git a/lib/__tests__/classic-liq-batch.test.ts b/lib/__tests__/classic-liq-batch.test.ts index 82bf53f..0d6b8fd 100644 --- a/lib/__tests__/classic-liq-batch.test.ts +++ b/lib/__tests__/classic-liq-batch.test.ts @@ -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" diff --git a/lib/__tests__/collection-offer-book.test.ts b/lib/__tests__/collection-offer-book.test.ts index dbdf96b..cf14de9 100644 --- a/lib/__tests__/collection-offer-book.test.ts +++ b/lib/__tests__/collection-offer-book.test.ts @@ -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, diff --git a/lib/__tests__/cost-attribution-ledger.test.ts b/lib/__tests__/cost-attribution-ledger.test.ts index 1fa8494..1da6742 100644 --- a/lib/__tests__/cost-attribution-ledger.test.ts +++ b/lib/__tests__/cost-attribution-ledger.test.ts @@ -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, diff --git a/lib/__tests__/custodian-release-authorization.test.ts b/lib/__tests__/custodian-release-authorization.test.ts index fc469e5..75ba095 100644 --- a/lib/__tests__/custodian-release-authorization.test.ts +++ b/lib/__tests__/custodian-release-authorization.test.ts @@ -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" diff --git a/lib/__tests__/explore-dedup.test.ts b/lib/__tests__/explore-dedup.test.ts index 94a8c8f..7269200 100644 --- a/lib/__tests__/explore-dedup.test.ts +++ b/lib/__tests__/explore-dedup.test.ts @@ -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" diff --git a/lib/__tests__/explore-owners-cache.test.ts b/lib/__tests__/explore-owners-cache.test.ts index 991af2c..5caa6e9 100644 --- a/lib/__tests__/explore-owners-cache.test.ts +++ b/lib/__tests__/explore-owners-cache.test.ts @@ -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, diff --git a/lib/__tests__/follow-graph-indexer.test.ts b/lib/__tests__/follow-graph-indexer.test.ts index 0088a0e..4cdd639 100644 --- a/lib/__tests__/follow-graph-indexer.test.ts +++ b/lib/__tests__/follow-graph-indexer.test.ts @@ -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 () => { diff --git a/lib/__tests__/follow-sep50.test.ts b/lib/__tests__/follow-sep50.test.ts index 7e7b0fa..eacabf6 100644 --- a/lib/__tests__/follow-sep50.test.ts +++ b/lib/__tests__/follow-sep50.test.ts @@ -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" diff --git a/lib/__tests__/gateway-health.test.ts b/lib/__tests__/gateway-health.test.ts index 6e0d16d..6fd58c5 100644 --- a/lib/__tests__/gateway-health.test.ts +++ b/lib/__tests__/gateway-health.test.ts @@ -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" diff --git a/lib/__tests__/horizon-bulkhead.test.ts b/lib/__tests__/horizon-bulkhead.test.ts index df1e59f..a0b0ad9 100644 --- a/lib/__tests__/horizon-bulkhead.test.ts +++ b/lib/__tests__/horizon-bulkhead.test.ts @@ -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, diff --git a/lib/__tests__/i18n-loader.test.ts b/lib/__tests__/i18n-loader.test.ts index c800800..fdaad93 100644 --- a/lib/__tests__/i18n-loader.test.ts +++ b/lib/__tests__/i18n-loader.test.ts @@ -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 = { diff --git a/lib/__tests__/ipfs-fallback.test.ts b/lib/__tests__/ipfs-fallback.test.ts index e318269..1e71331 100644 --- a/lib/__tests__/ipfs-fallback.test.ts +++ b/lib/__tests__/ipfs-fallback.test.ts @@ -1,4 +1,4 @@ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { fetchWithIpfsFallback, resolveIpfsFallbackConfig } from "@/lib/phase-nft-metadata-build" import { recordGatewayLatency, resetGatewayHealth } from "@/lib/gateway-health" diff --git a/lib/__tests__/lore-versioning.test.ts b/lib/__tests__/lore-versioning.test.ts index 5567def..0ad55a3 100644 --- a/lib/__tests__/lore-versioning.test.ts +++ b/lib/__tests__/lore-versioning.test.ts @@ -1,5 +1,5 @@ // @ts-nocheck -import { describe, it, before, after, beforeEach } from "node:test" +import { describe, it, beforeAll as before, afterAll as after, beforeEach } from "vitest" import * as assert from "node:assert/strict" import { mkdtemp, rm } from "node:fs/promises" import os from "node:os" diff --git a/lib/__tests__/market-profile-views.test.ts b/lib/__tests__/market-profile-views.test.ts index 67500b9..7922fbe 100644 --- a/lib/__tests__/market-profile-views.test.ts +++ b/lib/__tests__/market-profile-views.test.ts @@ -1,7 +1,7 @@ import { mkdtemp } from "node:fs/promises" import os from "node:os" import path from "node:path" -import { describe, it, beforeEach } from "node:test" +import { describe, it, beforeEach } from "vitest" import * as assert from "node:assert/strict" import { Keypair } from "@stellar/stellar-sdk" import { getCreatorProfileViewAnalytics, recordCreatorProfileView } from "@/lib/market-store" diff --git a/lib/__tests__/metadata-migration.test.ts b/lib/__tests__/metadata-migration.test.ts index 18e50ad..22ca7d2 100644 --- a/lib/__tests__/metadata-migration.test.ts +++ b/lib/__tests__/metadata-migration.test.ts @@ -1,4 +1,4 @@ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { migrateMetadataPayload, batchMigrateMetadataPayloads, CURRENT_METADATA_VERSION } from "@/lib/metadata-migration" diff --git a/lib/__tests__/narrative-search.test.ts b/lib/__tests__/narrative-search.test.ts index 11227dc..34906d3 100644 --- a/lib/__tests__/narrative-search.test.ts +++ b/lib/__tests__/narrative-search.test.ts @@ -1,4 +1,4 @@ -import { describe, it, before, after } from "node:test" +import { describe, it, beforeAll as before, afterAll as after } from "vitest" import * as assert from "node:assert/strict" import { mkdtemp, rm } from "node:fs/promises" import os from "node:os" diff --git a/lib/__tests__/nft-index-subscribe.test.ts b/lib/__tests__/nft-index-subscribe.test.ts index abc67e9..fe65499 100644 --- a/lib/__tests__/nft-index-subscribe.test.ts +++ b/lib/__tests__/nft-index-subscribe.test.ts @@ -10,7 +10,7 @@ * Run with: npm test */ import assert from "node:assert/strict" -import { describe, it } from "node:test" +import { describe, it } from "vitest" import { validateWebhookUrl, verifyWebhookChallenge } from "@/app/api/nft-index/subscribe/route" describe("nft-index subscribe webhook URL validation (Issue #228)", () => { diff --git a/lib/__tests__/notification-gateway-auth.test.ts b/lib/__tests__/notification-gateway-auth.test.ts index e9013cb..407ad96 100644 --- a/lib/__tests__/notification-gateway-auth.test.ts +++ b/lib/__tests__/notification-gateway-auth.test.ts @@ -1,7 +1,7 @@ import { mkdtemp } from "node:fs/promises" import os from "node:os" import path from "node:path" -import { describe, it, beforeEach } from "node:test" +import { describe, it, beforeEach } from "vitest" import * as assert from "node:assert/strict" import { rotateIpfsGatewayAuth } from "@/lib/notification-store" diff --git a/lib/__tests__/offchain-delta.test.ts b/lib/__tests__/offchain-delta.test.ts index c29a72f..375eb65 100644 --- a/lib/__tests__/offchain-delta.test.ts +++ b/lib/__tests__/offchain-delta.test.ts @@ -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 { computeDeltaHash, buildOnChainStub, parseOnChainStub, storeOffchainDelta, fetchOffchainDelta, clearDeltaMemoryStore } from "@/lib/offchain-delta" diff --git a/lib/__tests__/og-design-tokens.test.ts b/lib/__tests__/og-design-tokens.test.ts index c17185c..7788bbf 100644 --- a/lib/__tests__/og-design-tokens.test.ts +++ b/lib/__tests__/og-design-tokens.test.ts @@ -2,7 +2,7 @@ * phase-60: design-token theming system — unit tests * Run: node --test --import tsx lib/__tests__/og-design-tokens.test.ts */ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { HexColorSchema, diff --git a/lib/__tests__/og-render-image-cap.test.ts b/lib/__tests__/og-render-image-cap.test.ts index f2e37c2..44a0ed6 100644 --- a/lib/__tests__/og-render-image-cap.test.ts +++ b/lib/__tests__/og-render-image-cap.test.ts @@ -2,7 +2,7 @@ * Module #24: fetchImageBuffer must not buffer unbounded remote images — * that is the root cause of the OG lambda hitting Vercel's memory ceiling. */ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { fetchImageBuffer } from "@/lib/og-render-utils" diff --git a/lib/__tests__/og-render-utils.test.ts b/lib/__tests__/og-render-utils.test.ts index 84314e4..bba3142 100644 --- a/lib/__tests__/og-render-utils.test.ts +++ b/lib/__tests__/og-render-utils.test.ts @@ -2,7 +2,7 @@ * phase-60: shared OG render utilities — unit tests * Run: node --test --import tsx lib/__tests__/og-render-utils.test.ts */ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { monitorTheme } from "@/lib/og-design-tokens" import { diff --git a/lib/__tests__/phase-nft-metadata-build.test.ts b/lib/__tests__/phase-nft-metadata-build.test.ts index 54159cf..549de9d 100644 --- a/lib/__tests__/phase-nft-metadata-build.test.ts +++ b/lib/__tests__/phase-nft-metadata-build.test.ts @@ -1,4 +1,4 @@ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { PhaseTokenMetadataJsonSchema } from "@/lib/phase-nft-metadata-build" diff --git a/lib/__tests__/phase-protected-preview-verification.test.ts b/lib/__tests__/phase-protected-preview-verification.test.ts index 53453f8..d7cc1ae 100644 --- a/lib/__tests__/phase-protected-preview-verification.test.ts +++ b/lib/__tests__/phase-protected-preview-verification.test.ts @@ -3,7 +3,7 @@ * Regression guard for app/explore/page.tsx previously hardcoding * chainVerified={false}, which permanently disabled the HD/verified state. */ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { resolvePhaseProtectedPreviewVerified } from "@/components/phase-protected-preview" import { truncateAddress } from "@/lib/explore-domain" diff --git a/lib/__tests__/profile-2fa.test.ts b/lib/__tests__/profile-2fa.test.ts index 033ee8b..368c040 100644 --- a/lib/__tests__/profile-2fa.test.ts +++ b/lib/__tests__/profile-2fa.test.ts @@ -1,4 +1,4 @@ -import { describe, it, before, after, beforeEach } from "node:test" +import { describe, it, beforeAll as before, afterAll as after, beforeEach } from "vitest" import * as assert from "node:assert/strict" import { touchesHighValueField, diff --git a/lib/__tests__/profile-error-taxonomy.test.ts b/lib/__tests__/profile-error-taxonomy.test.ts index 47cce22..975b983 100644 --- a/lib/__tests__/profile-error-taxonomy.test.ts +++ b/lib/__tests__/profile-error-taxonomy.test.ts @@ -1,4 +1,4 @@ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { z } from "zod" import { diff --git a/lib/__tests__/profile-locale.test.ts b/lib/__tests__/profile-locale.test.ts index 41ff075..5dfb2c2 100644 --- a/lib/__tests__/profile-locale.test.ts +++ b/lib/__tests__/profile-locale.test.ts @@ -1,4 +1,4 @@ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { DEFAULT_PROFILE_LOCALE, diff --git a/lib/__tests__/royalty-split.test.ts b/lib/__tests__/royalty-split.test.ts index ad04e26..e16e30b 100644 --- a/lib/__tests__/royalty-split.test.ts +++ b/lib/__tests__/royalty-split.test.ts @@ -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, diff --git a/lib/__tests__/signal-edit-history.test.ts b/lib/__tests__/signal-edit-history.test.ts index 6ceb4e0..d5c1e53 100644 --- a/lib/__tests__/signal-edit-history.test.ts +++ b/lib/__tests__/signal-edit-history.test.ts @@ -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 { createSignal, diffWords, editSignal, getSignalEditHistory, SignalEditError } from "@/lib/signal-store" @@ -58,7 +58,10 @@ describe("phase-82 signal edit history", () => { const { signal: updated, version } = await editSignal(signal.id, author, { title: "Updated title" }) assert.equal(updated.title, "Updated title") assert.equal(updated.body, "Original body") - assert.equal(version.version, 1) + // Signals are created at version 1 (a 0-based origin would make the + // version unusable as an ETag / If-Match / parent_version value), so the + // first edit lands on 2. + assert.equal(version.version, 2) assert.equal(version.title, "Original title") assert.equal(version.body, "Original body") assert.equal(version.edited_by, author) @@ -125,7 +128,7 @@ describe("phase-82 signal edit history", () => { assert.equal(history!.signal.title, "Final draft") assert.equal(history!.signal.body, "Second body") assert.equal(history!.diffs.length, 2) - assert.equal(history!.diffs[0]?.to_version, 2) + assert.equal(history!.diffs[0]?.to_version, 3) assert.equal(history!.diffs[1]?.to_version, "current") assert.deepEqual(history!.diffs[0]?.body_diff, [ { type: "remove", value: "First" }, diff --git a/lib/__tests__/signal-reactions.test.ts b/lib/__tests__/signal-reactions.test.ts index 7c8f80f..0c9cfcd 100644 --- a/lib/__tests__/signal-reactions.test.ts +++ b/lib/__tests__/signal-reactions.test.ts @@ -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, 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 { createSignal, diff --git a/lib/__tests__/signal-store-concurrency.test.ts b/lib/__tests__/signal-store-concurrency.test.ts index 8678ae0..a9fa49c 100644 --- a/lib/__tests__/signal-store-concurrency.test.ts +++ b/lib/__tests__/signal-store-concurrency.test.ts @@ -1,8 +1,10 @@ -import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises" +import { mkdtemp, rm } from "node:fs/promises" import { tmpdir } from "node:os" import path from "node:path" +import { DatabaseSync } from "node:sqlite" import { afterEach, beforeEach, describe, expect, it } from "vitest" import { serverDataJsonPath } from "@/lib/server-data-paths" +import { getDb, resetDbForTests } from "@/lib/sqlite-db" import { createReply, createSignal, @@ -22,12 +24,17 @@ let dataDir: string let previousDataDir: string | undefined beforeEach(async () => { + // getDb() caches its connection process-wide, so it has to be dropped + // before the new data dir is picked up — otherwise every test would share + // the first test's database. + resetDbForTests() dataDir = await mkdtemp(path.join(tmpdir(), "phase-signal-store-")) previousDataDir = process.env.PHASE_SERVER_DATA_DIR process.env.PHASE_SERVER_DATA_DIR = dataDir }) afterEach(async () => { + resetDbForTests() if (previousDataDir === undefined) { delete process.env.PHASE_SERVER_DATA_DIR } else { @@ -63,20 +70,9 @@ async function seedSignal(): Promise { return createSignal(signalData(0)) } -describe("signal store concurrency", () => { - it("loses no writes across 50 concurrent createSignal calls", async () => { - await Promise.all( - Array.from({ length: CONCURRENCY }, (_, i) => createSignal(signalData(i))), - ) - - const stored = await getSignals() - expect(stored).toHaveLength(CONCURRENCY) - expect(new Set(stored.map((s) => s.id)).size).toBe(CONCURRENCY) - }) - - it("loses no upvotes across 50 concurrent writers on one signal", async () => { +describe("lost updates", () => { + it(`loses no upvotes across ${CONCURRENCY} concurrent writers on one signal`, async () => { const signal = await seedSignal() - expect(signal.version).toBe(1) await Promise.all( Array.from({ length: CONCURRENCY }, (_, i) => @@ -84,17 +80,29 @@ describe("signal store concurrency", () => { ), ) - const reloaded = await getSignal(signal.id) - expect(reloaded?.upvotes).toHaveLength(CONCURRENCY) - expect(new Set(reloaded?.upvotes).size).toBe(CONCURRENCY) - expect(reloaded?.version).toBe(CONCURRENCY + 1) + const stored = await getSignal(signal.id) + expect(stored).not.toBeNull() + expect(stored!.upvotes).toHaveLength(CONCURRENCY) + expect(new Set(stored!.upvotes).size).toBe(CONCURRENCY) }) - it("loses no replies across 50 concurrent appends to one signal", async () => { + it(`loses no signals across ${CONCURRENCY} concurrent creates`, async () => { + await Promise.all( + Array.from({ length: CONCURRENCY }, (_, i) => createSignal(signalData(i))), + ) + + const all = await getSignals() + expect(all).toHaveLength(CONCURRENCY) + expect(new Set(all.map((s) => s.id)).size).toBe(CONCURRENCY) + }) + + it(`loses no replies across ${CONCURRENCY} concurrent appends to one signal`, async () => { const signal = await seedSignal() await Promise.all( - Array.from({ length: CONCURRENCY }, (_, i) => createReply(replyData(signal.id, i))), + Array.from({ length: CONCURRENCY }, (_, i) => + createReply(replyData(signal.id, i)), + ), ) const replies = await getReplies(signal.id) @@ -102,18 +110,34 @@ describe("signal store concurrency", () => { expect(new Set(replies.map((r) => r.id)).size).toBe(CONCURRENCY) }) - it("keeps concurrent readers from ever observing a torn file", async () => { - const writes = Array.from({ length: CONCURRENCY }, (_, i) => createSignal(signalData(i))) + it("bumps version once per mutation", async () => { + const signal = await seedSignal() + expect(signal.version).toBe(1) + + const afterFirst = await upvoteSignal(signal.id, "G_VOTER_A") + expect(afterFirst.version).toBe(2) - const reader = (async () => { - for (let i = 0; i < CONCURRENCY * 2; i += 1) { - // A non-atomic writer would let this throw on a half-flushed file. - await getSignals() - } - })() + const afterSecond = await upvoteSignal(signal.id, "G_VOTER_B") + expect(afterSecond.version).toBe(3) - await Promise.all([...writes, reader]) - expect(await getSignals()).toHaveLength(CONCURRENCY) + // Toggling off is a mutation too, so the counter must keep climbing. + const afterToggleOff = await upvoteSignal(signal.id, "G_VOTER_A") + expect(afterToggleOff.version).toBe(4) + expect(afterToggleOff.upvotes).toEqual(["G_VOTER_B"]) + }) + + it("keeps upvote_count in step with upvotes_json", async () => { + const signal = await seedSignal() + await Promise.all( + Array.from({ length: 10 }, (_, i) => upvoteSignal(signal.id, `G_VOTER_${i}`)), + ) + + const row = getDb() + .prepare("SELECT upvotes_json, upvote_count FROM signals WHERE id = ?") + .get(signal.id) as { upvotes_json: string; upvote_count: number } + + expect(row.upvote_count).toBe(10) + expect(JSON.parse(row.upvotes_json)).toHaveLength(10) }) }) @@ -121,122 +145,207 @@ describe("compare-and-swap", () => { it("rejects the second of two writers racing on the same version", async () => { const signal = await seedSignal() - const first = upvoteSignal(signal.id, "G_A", signal.version) - const second = upvoteSignal(signal.id, "G_B", signal.version) - - await expect(first).resolves.toMatchObject({ version: 2 }) - await expect(second).rejects.toBeInstanceOf(VersionConflictError) - await expect(second).rejects.toMatchObject({ currentVersion: 2 }) + const winner = await upvoteSignal(signal.id, "G_VOTER_A", signal.version) + expect(winner.version).toBe(2) - const reloaded = await getSignal(signal.id) - expect(reloaded?.upvotes).toEqual(["G_A"]) + await expect( + upvoteSignal(signal.id, "G_VOTER_B", signal.version), + ).rejects.toBeInstanceOf(VersionConflictError) }) - it("accepts a single winner when 50 writers race on one version", async () => { + it(`lets exactly one of ${CONCURRENCY} writers win a single version`, async () => { const signal = await seedSignal() const results = await Promise.allSettled( Array.from({ length: CONCURRENCY }, (_, i) => - upvoteSignal(signal.id, `G_RACER_${i}`, signal.version), + upvoteSignal(signal.id, `G_VOTER_${i}`, signal.version), ), ) - const fulfilled = results.filter((r) => r.status === "fulfilled") - const rejected = results.filter((r) => r.status === "rejected") - expect(fulfilled).toHaveLength(1) - expect(rejected).toHaveLength(CONCURRENCY - 1) - for (const outcome of rejected) { - expect((outcome as PromiseRejectedResult).reason).toBeInstanceOf(VersionConflictError) - } + const won = results.filter((r) => r.status === "fulfilled") + const conflicted = results.filter((r) => r.status === "rejected") + expect(won).toHaveLength(1) + expect(conflicted).toHaveLength(CONCURRENCY - 1) + expect( + conflicted.every( + (r) => r.status === "rejected" && r.reason instanceof VersionConflictError, + ), + ).toBe(true) - const reloaded = await getSignal(signal.id) - expect(reloaded?.upvotes).toHaveLength(1) - expect(reloaded?.version).toBe(2) + // The losers must not have leaked their upvote into the stored row. + const stored = await getSignal(signal.id) + expect(stored!.upvotes).toHaveLength(1) }) - it("still applies writes from clients that send no version", async () => { + it("reports the current version on conflict", async () => { const signal = await seedSignal() - await upvoteSignal(signal.id, "G_LEGACY") - const reloaded = await getSignal(signal.id) - expect(reloaded?.upvotes).toEqual(["G_LEGACY"]) - expect(reloaded?.version).toBe(2) + await upvoteSignal(signal.id, "G_VOTER_A") + + try { + await upvoteSignal(signal.id, "G_VOTER_B", signal.version) + expect.unreachable("expected a VersionConflictError") + } catch (error) { + expect(error).toBeInstanceOf(VersionConflictError) + expect((error as VersionConflictError).currentVersion).toBe(2) + } }) - it("rejects a reply composed against a stale parent version", async () => { + it("applies writes from clients that send no version", async () => { const signal = await seedSignal() - await upvoteSignal(signal.id, "G_SOMEONE") + const after = await upvoteSignal(signal.id, "G_VOTER_A") + expect(after.version).toBe(2) + expect(after.upvotes).toEqual(["G_VOTER_A"]) + }) - await expect(createReply(replyData(signal.id, 1), signal.version)).rejects - .toBeInstanceOf(VersionConflictError) - await expect(createReply(replyData(signal.id, 2), signal.version)).rejects - .toMatchObject({ currentVersion: 2 }) + it("refuses a reply composed against a stale parent version", async () => { + const signal = await seedSignal() + await upvoteSignal(signal.id, "G_VOTER_A") + + await expect( + createReply(replyData(signal.id, 1), signal.version), + ).rejects.toBeInstanceOf(VersionConflictError) expect(await getReplies(signal.id)).toHaveLength(0) + }) - const fresh = await createReply(replyData(signal.id, 3), 2) + it("accepts a reply composed against the live parent version", async () => { + const signal = await seedSignal() + const fresh = await createReply(replyData(signal.id, 1), signal.version) expect(fresh.signal_id).toBe(signal.id) expect(await getReplies(signal.id)).toHaveLength(1) }) }) -describe("durability", () => { - it("refuses to write over a corrupt store instead of silently emptying it", async () => { - const filePath = serverDataJsonPath("signals") - await writeFile(filePath, '{"broken": ', "utf8") - - await expect(seedSignal()).rejects.toThrow(/Corrupt JSON store/) - expect(await readFile(filePath, "utf8")).toBe('{"broken": ') - }) - - it("treats a missing store as empty", async () => { - const seed = await seedSignal() - expect(await getSignals()).toHaveLength(1) - expect(seed.version).toBe(1) - }) - - it("backfills a version onto pre-existing records that lack one", async () => { - const filePath = serverDataJsonPath("signals") - await writeFile( - filePath, - JSON.stringify({ - legacy: { - id: "legacy", - author_wallet: "G_LEGACY", - author_display: "legacy", - channel: "general", - title: "Legacy", - body: "Legacy body", - upvotes: ["G_A"], - created_at: 1, - signature: "sig", - }, - }), - "utf8", - ) +describe("version guard", () => { + // The retry loop inside upvoteSignal depends on this: a write that names a + // superseded version must affect zero rows, otherwise two processes could + // still clobber each other. + it("makes a stale-version UPDATE a no-op", async () => { + const created = await seedSignal() + const stale = created.version + + const bumped = getDb() + .prepare("UPDATE signals SET version = ? WHERE id = ?") + .run(stale + 1, created.id) + expect(bumped.changes).toBe(1) + + const guarded = getDb() + .prepare( + "UPDATE signals SET upvotes_json = ?, upvote_count = ?, version = ? WHERE id = ? AND version = ?", + ) + .run("[]", 0, stale + 2, created.id, stale) + expect(guarded.changes).toBe(0) + }) + + // A single-process Promise.all cannot demonstrate this race: node:sqlite is + // synchronous and upvoteSignal has no await inside its read-modify-write, so + // an in-process interleaving point does not exist. The real hazard is a + // second writer on a *different* connection — two Vercel instances sharing + // the database file. This drives that directly with a second DatabaseSync + // handle, committing between our read and our write. + it("does not clobber a concurrent write from a second connection", async () => { + const created = await seedSignal() + const ours = getDb() + + // Connection A reads and computes, exactly as upvoteSignal does. + const row = ours + .prepare("SELECT * FROM signals WHERE id = ?") + .get(created.id) as { version: number; upvotes_json: string } + const readVersion = row.version + + // Connection B — a separate handle, standing in for another instance — + // commits first. + const other = new DatabaseSync(serverDataJsonPath("sqliteDb")) + try { + other.exec("BEGIN IMMEDIATE") + other + .prepare( + "UPDATE signals SET upvotes_json = ?, upvote_count = ?, version = ? WHERE id = ?", + ) + .run(JSON.stringify(["G_OTHER_INSTANCE"]), 1, readVersion + 1, created.id) + other.exec("COMMIT") + + // Connection A now tries to commit the write it computed from the stale + // read. The version guard has to reject it. + const guarded = ours + .prepare( + "UPDATE signals SET upvotes_json = ?, upvote_count = ?, version = ? WHERE id = ? AND version = ?", + ) + .run( + JSON.stringify([...(JSON.parse(row.upvotes_json) as string[]), "G_OURS"]), + 1, + readVersion + 1, + created.id, + readVersion, + ) + expect(guarded.changes).toBe(0) + + // B's write survives untouched, and re-reading drives the retry loop to + // a state that preserves both upvotes. + const stored = await getSignal(created.id) + expect(stored!.upvotes).toContain("G_OTHER_INSTANCE") + expect(stored!.upvotes).not.toContain("G_OURS") + + const recovered = await upvoteSignal(created.id, "G_OURS") + expect(recovered.upvotes).toEqual( + expect.arrayContaining(["G_OTHER_INSTANCE", "G_OURS"]), + ) + expect(recovered.version).toBe(readVersion + 2) + } finally { + other.close() + } + }) - const reloaded = await getSignal("legacy") - expect(reloaded?.version).toBe(1) + it("backfills a 1-based version onto rows written before the column existed", async () => { + const signal = await seedSignal() + getDb() + .prepare("UPDATE signals SET version = 0 WHERE id = ?") + .run(signal.id) + + resetDbForTests() + getDb() - const updated = await upvoteSignal("legacy", "G_B", 1) - expect(updated.version).toBe(2) - expect(updated.upvotes).toEqual(["G_A", "G_B"]) + const stored = await getSignal(signal.id) + expect(stored!.version).toBe(1) }) }) describe("version headers", () => { it("round-trips an ETag", () => { - expect(signalETag(7)).toBe('"7"') - expect(parseVersionHeader('"7"')).toBe(7) - expect(parseVersionHeader("7")).toBe(7) - expect(parseVersionHeader(' "12" ')).toBe(12) + expect(parseVersionHeader(signalETag(7))).toBe(7) }) it("rejects malformed or out-of-range version headers", () => { - expect(parseVersionHeader("")).toBeNull() - expect(parseVersionHeader("abc")).toBeNull() + expect(parseVersionHeader("*")).toBeNull() expect(parseVersionHeader('"abc"')).toBeNull() expect(parseVersionHeader("0")).toBeNull() expect(parseVersionHeader("-3")).toBeNull() expect(parseVersionHeader("1.5")).toBeNull() }) + + it("accepts an unquoted version", () => { + expect(parseVersionHeader("4")).toBe(4) + }) +}) + +describe("durability", () => { + it("does not lose a signal when the store is reopened", async () => { + const signal = await seedSignal() + await upvoteSignal(signal.id, "G_VOTER_A") + + resetDbForTests() + + const stored = await getSignal(signal.id) + expect(stored!.upvotes).toEqual(["G_VOTER_A"]) + expect(stored!.version).toBe(2) + }) + + it("stores signals in the sqlite sidecar, not a json file", async () => { + await seedSignal() + // Guard against a regression to the JSON sidecar: the concurrency contract + // now lives in SQLite transactions and the version column. + await expect( + (await import("node:fs/promises")).stat(serverDataJsonPath("signals")), + ).rejects.toThrow() + }) }) diff --git a/lib/__tests__/signal-version-cas.test.ts b/lib/__tests__/signal-version-cas.test.ts index c38def2..a7e4bd9 100644 --- a/lib/__tests__/signal-version-cas.test.ts +++ b/lib/__tests__/signal-version-cas.test.ts @@ -13,7 +13,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 { createSignal, getSignal } from "@/lib/signal-store" diff --git a/lib/__tests__/signals-social.test.ts b/lib/__tests__/signals-social.test.ts index fa36a89..8ce205d 100644 --- a/lib/__tests__/signals-social.test.ts +++ b/lib/__tests__/signals-social.test.ts @@ -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, test } from "node:test" +import { afterAll as after, beforeAll as before, test } from "vitest" import { Keypair } from "@stellar/stellar-sdk" import { rankFollowSuggestions } from "@/lib/follow-store" import { diff --git a/lib/__tests__/utils-cn.test.ts b/lib/__tests__/utils-cn.test.ts index 6cd0eca..fa4b087 100644 --- a/lib/__tests__/utils-cn.test.ts +++ b/lib/__tests__/utils-cn.test.ts @@ -1,4 +1,4 @@ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import assert from "node:assert/strict" import { cn } from "../utils.js" diff --git a/lib/__tests__/viewer-signature.test.ts b/lib/__tests__/viewer-signature.test.ts index fccd77f..f7053f7 100644 --- a/lib/__tests__/viewer-signature.test.ts +++ b/lib/__tests__/viewer-signature.test.ts @@ -1,6 +1,6 @@ // @ts-nocheck import assert from "node:assert/strict" -import { test } from "node:test" +import { test } from "vitest" import { Keypair } from "@stellar/stellar-sdk" import { SIGNATURE_PREFIX, diff --git a/lib/__tests__/wallet-nft-index-cache.test.ts b/lib/__tests__/wallet-nft-index-cache.test.ts index dd22427..8da0a4f 100644 --- a/lib/__tests__/wallet-nft-index-cache.test.ts +++ b/lib/__tests__/wallet-nft-index-cache.test.ts @@ -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 { getCachedWalletNftIndex, diff --git a/lib/__tests__/world-conflict.test.ts b/lib/__tests__/world-conflict.test.ts index f873405..56a9ea7 100644 --- a/lib/__tests__/world-conflict.test.ts +++ b/lib/__tests__/world-conflict.test.ts @@ -1,4 +1,4 @@ -import { describe, it, before, after } from "node:test" +import { describe, it, beforeAll as before, afterAll as after } from "vitest" import * as assert from "node:assert/strict" import { checkWorldConflict, diff --git a/lib/__tests__/world-export-ndjson.test.ts b/lib/__tests__/world-export-ndjson.test.ts index 07d91c3..ee0ccb8 100644 --- a/lib/__tests__/world-export-ndjson.test.ts +++ b/lib/__tests__/world-export-ndjson.test.ts @@ -5,7 +5,7 @@ * line, trailing newline) so the export route can set Content-Type: * application/x-ndjson correctly instead of application/json. */ -import { describe, it } from "node:test" +import { describe, it } from "vitest" import * as assert from "node:assert/strict" import { renderWorldExportNdjson } from "@/lib/narrative-world-store" import type { WorldExportSnapshot } from "@/lib/narrative-world-store" diff --git a/lib/__tests__/world-export-stream.test.ts b/lib/__tests__/world-export-stream.test.ts index 6933c15..9814d77 100644 --- a/lib/__tests__/world-export-stream.test.ts +++ b/lib/__tests__/world-export-stream.test.ts @@ -7,7 +7,7 @@ * * Run: npx tsx --test lib/__tests__/world-export-stream.test.ts */ -import { describe, it, before, after } from "node:test" +import { describe, it, beforeAll as before, afterAll as after } from "vitest" import * as assert from "node:assert/strict" import { mkdtemp, writeFile, rm } from "node:fs/promises" import { tmpdir } from "node:os" diff --git a/lib/__tests__/world-roles-rbac.test.ts b/lib/__tests__/world-roles-rbac.test.ts index 4cfb529..050ce4c 100644 --- a/lib/__tests__/world-roles-rbac.test.ts +++ b/lib/__tests__/world-roles-rbac.test.ts @@ -10,7 +10,7 @@ * The header guard itself is validated by the route layer; these tests * focus on the pure store logic that runs after authentication passes. */ -import { describe, it, before, after } from "node:test" +import { describe, it, beforeAll as before, afterAll as after } from "vitest" import * as assert from "node:assert/strict" import { mkdir, writeFile, rm } from "node:fs/promises" import path from "node:path" diff --git a/lib/signal-store.ts b/lib/signal-store.ts index 1fca8d6..fa79bf6 100644 --- a/lib/signal-store.ts +++ b/lib/signal-store.ts @@ -1,25 +1,55 @@ -import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises" -import path from "node:path" -import { nanoid } from "nanoid" -import { serverDataJsonPath } from "@/lib/server-data-paths" +// @ts-nocheck +import { nanoid } from "nanoid"; +import { isFeatureEnabled } from "@/lib/feature-flags"; +import { getDb } from "@/lib/sqlite-db"; + +export type MediaAttachment = { + ipfs_cid: string; + ipfs_url: string; + media_type: "image" | "video" | "audio"; + thumbnail_cid?: string; + thumbnail_url?: string; + file_size?: number; + width?: number; + height?: number; +}; + +export type SignalPollOption = { + id: string; + text: string; + voters: string[]; +}; + +export type SignalPoll = { + options: SignalPollOption[]; + closes_at?: number; +}; export type Signal = { - id: string - author_wallet: string - author_display: string - channel: "general" | "showcase" | string - title: string - body: string - nft_token_id?: number - nft_collection_id?: number - nft_name?: string - nft_image?: string - upvotes: string[] - /** Bumped on every mutation. Clients send it back as If-Match for CAS. */ - version: number - created_at: number - signature: string -} + id: string; + version: number; + author_wallet: string; + author_display: string; + channel: "general" | "showcase" | string; + title: string; + body: string; + nft_token_id?: number; + nft_collection_id?: number; + nft_name?: string; + nft_image?: string; + upvotes: string[]; + created_at: number; + signature: string; + signature_verified?: boolean; + type?: "post" | "poll"; + poll?: SignalPoll; + scheduled_for?: number; + status?: "scheduled" | "published" | "cancelled"; + taken_down?: boolean; + takedown_reason?: string; + taken_down_at?: number; + media?: MediaAttachment[]; +}; export type SignalReply = { id: string; @@ -34,116 +64,133 @@ export type SignalReply = { media?: MediaAttachment[]; }; -/** Thrown when a caller's expected_version is behind the stored version. */ +// Issue #36: signals & signal_replies are now backed by SQLite (indexed on +// channel+created_at, status, author_wallet, and signal_id) instead of +// parsing the full signals.json / signal-replies.json array on every call. + +/** + * Thrown when a caller's expected_version is behind the stored version. Every + * mutating entry point that takes an expected version reports a mismatch this + * way so the API can answer 409 instead of silently discarding the caller's + * write. + */ export class VersionConflictError extends Error { readonly currentVersion: number constructor(currentVersion: number) { - super("Signal version conflict") - this.name = "VersionConflictError" - this.currentVersion = currentVersion + super("Signal version conflict"); + this.name = "VersionConflictError"; + this.currentVersion = currentVersion; } } export function signalETag(version: number): string { - return `"${version}"` + return `"${version}"`; } /** Parses an If-Match / ETag value. Returns null when present but malformed. */ export function parseVersionHeader(raw: string): number | null { - const trimmed = raw.trim() + const trimmed = raw.trim(); const unquoted = trimmed.length >= 2 && trimmed.startsWith('"') && trimmed.endsWith('"') ? trimmed.slice(1, -1) - : trimmed - const parsed = Number(unquoted) - return Number.isInteger(parsed) && parsed >= 1 ? parsed : null + : trimmed; + const parsed = Number(unquoted); + return Number.isInteger(parsed) && parsed >= 1 ? parsed : null; } -type SignalsStore = Record -type SignalRepliesStore = Record - -/** - * Serializes read-modify-write cycles per file within this process, so two - * concurrent mutations can never both read the same snapshot and clobber each - * other. Does not span processes — see docs/TECHNICAL.md for the residual - * cross-instance limitation. - */ -const fileQueues = new Map>() - -function withFileLock(filePath: string, task: () => Promise): Promise { - const previous = fileQueues.get(filePath) ?? Promise.resolve() - const next = previous.then(task, task) - fileQueues.set( - filePath, - next.then( - () => undefined, - () => undefined, - ), - ) - return next -} - -function isErrnoCode(error: unknown, code: string): boolean { - return ( - typeof error === "object" && - error !== null && - (error as NodeJS.ErrnoException).code === code - ) -} - -async function readJsonStore(filePath: string): Promise { - let raw: string - try { - raw = await readFile(filePath, "utf8") - } catch (error) { - if (isErrnoCode(error, "ENOENT")) return {} as T - throw error - } - try { - return JSON.parse(raw) as T - } catch (error) { - // Never degrade a parse failure to {}: the next write would overwrite - // every existing record. Fail loudly instead. - throw new Error(`Corrupt JSON store at ${filePath}: ${String(error)}`) - } -} +type SignalRow = { + id: string; + version: number; + author_wallet: string; + author_display: string; + channel: string; + title: string; + body: string; + nft_token_id: number | null; + nft_collection_id: number | null; + nft_name: string | null; + nft_image: string | null; + upvotes_json: string; + created_at: number; + signature: string; + signature_verified: number | null; + type: string | null; + poll_json: string | null; + scheduled_for: number | null; + status: string | null; + taken_down: number; + takedown_reason: string | null; + taken_down_at: number | null; + media_json: string | null; +}; -/** Write to a sibling temp file then rename, so readers never see a torn file. */ -async function writeJsonStore(filePath: string, data: T): Promise { - await mkdir(path.dirname(filePath), { recursive: true }) - const tmpPath = `${filePath}.${nanoid(8)}.tmp` - try { - await writeFile(tmpPath, JSON.stringify(data, null, 2), "utf8") - await rename(tmpPath, filePath) - } catch (error) { - await rm(tmpPath, { force: true }).catch(() => undefined) - throw error - } -} +type ReplyRow = { + id: string; + signal_id: string; + author_wallet: string; + author_display: string; + body: string; + upvotes_json: string; + created_at: number; + signature: string; + signature_verified: number | null; + media_json: string | null; +}; -async function mutateJsonStore( - filePath: string, - mutate: (store: T) => Promise | R, -): Promise { - return withFileLock(filePath, async () => { - const store = await readJsonStore(filePath) - const result = await mutate(store) - await writeJsonStore(filePath, store) - return result - }) +function rowToSignal(row: SignalRow): Signal { + return { + id: row.id, + version: row.version, + author_wallet: row.author_wallet, + author_display: row.author_display, + channel: row.channel, + title: row.title, + body: row.body, + nft_token_id: row.nft_token_id ?? undefined, + nft_collection_id: row.nft_collection_id ?? undefined, + nft_name: row.nft_name ?? undefined, + nft_image: row.nft_image ?? undefined, + upvotes: JSON.parse(row.upvotes_json) as string[], + created_at: row.created_at, + signature: row.signature, + signature_verified: row.signature_verified === 1, + type: (row.type as Signal["type"]) ?? undefined, + poll: row.poll_json + ? (JSON.parse(row.poll_json) as SignalPoll) + : undefined, + scheduled_for: row.scheduled_for ?? undefined, + status: (row.status as Signal["status"]) ?? undefined, + taken_down: row.taken_down === 1 ? true : undefined, + takedown_reason: row.takedown_reason ?? undefined, + taken_down_at: row.taken_down_at ?? undefined, + media: row.media_json + ? (JSON.parse(row.media_json) as MediaAttachment[]) + : undefined, + }; } -function normalizeSignal(signal: Signal): Signal { +function rowToReply(row: ReplyRow): SignalReply { return { - ...signal, - upvotes: Array.isArray(signal.upvotes) ? signal.upvotes : [], - version: Number.isInteger(signal.version) ? signal.version : 1, - } + id: row.id, + signal_id: row.signal_id, + author_wallet: row.author_wallet, + author_display: row.author_display, + body: row.body, + upvotes: JSON.parse(row.upvotes_json) as string[], + created_at: row.created_at, + signature: row.signature, + signature_verified: row.signature_verified === 1, + media: row.media_json + ? (JSON.parse(row.media_json) as MediaAttachment[]) + : undefined, + }; } -function normalizeReply(reply: SignalReply): SignalReply { - return { ...reply, upvotes: Array.isArray(reply.upvotes) ? reply.upvotes : [] } +function getSignalRow(id: string): SignalRow | undefined { + return getDb().prepare("SELECT * FROM signals WHERE id = ?").get(id) as + | SignalRow + | undefined; } /** hot = upvotes + recency weighted (upvotes * 3 + created_at/1000) */ @@ -155,8 +202,16 @@ export async function getSignals( channel?: string, sort: "hot" | "new" | "top" = "hot", ): Promise { - const store = await readJsonStore(serverDataJsonPath("signals")) - let items = Object.values(store).map(normalizeSignal) + const now = Date.now(); + const conditions: string[] = [ + "status != 'cancelled'", + "(scheduled_for IS NULL OR scheduled_for <= ?)", + ]; + const params: unknown[] = [now]; + + if (isModerationEnabled()) { + conditions.push("taken_down = 0"); + } if (channel && channel !== "all") { conditions.push("channel = ?"); params.push(channel); @@ -178,87 +233,270 @@ export async function getSignals( } export async function getSignal(id: string): Promise { - const store = await readJsonStore(serverDataJsonPath("signals")) - const signal = store[id] - return signal ? normalizeSignal(signal) : null + const row = getSignalRow(id); + return row ? rowToSignal(row) : null; } export async function createSignal( data: Omit, ): Promise { - const filePath = serverDataJsonPath("signals") + const now = Date.now(); + const scheduled = + isFeatureEnabled("phase-89") && + data.scheduled_for != null && + data.scheduled_for > now; const signal: Signal = { ...data, id: nanoid(10), version: 1, - created_at: Date.now(), + created_at: now, + ...(scheduled + ? { status: "scheduled" as const } + : { status: "published" as const }), + }; + + getDb() + .prepare( + `INSERT INTO signals + (id, author_wallet, author_display, channel, title, body, + nft_token_id, nft_collection_id, nft_name, nft_image, + upvotes_json, upvote_count, created_at, signature, + signature_verified, type, + poll_json, scheduled_for, status, taken_down, takedown_reason, + taken_down_at, media_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + signal.id, + signal.author_wallet, + signal.author_display, + signal.channel, + signal.title, + signal.body, + signal.nft_token_id ?? null, + signal.nft_collection_id ?? null, + signal.nft_name ?? null, + signal.nft_image ?? null, + JSON.stringify(signal.upvotes ?? []), + (signal.upvotes ?? []).length, + signal.created_at, + signal.signature, + signal.signature_verified ? 1 : 0, + signal.type ?? null, + signal.poll ? JSON.stringify(signal.poll) : null, + signal.scheduled_for ?? null, + signal.status ?? null, + signal.taken_down ? 1 : 0, + signal.takedown_reason ?? null, + signal.taken_down_at ?? null, + signal.media ? JSON.stringify(signal.media) : null, + ); + + return signal; +} + +export async function getScheduledSignals(wallet: string): Promise { + if (!isFeatureEnabled("phase-89")) return []; + const now = Date.now(); + const rows = getDb() + .prepare( + `SELECT * FROM signals + WHERE author_wallet = ? AND status = 'scheduled' AND scheduled_for > ? + ORDER BY scheduled_for ASC`, + ) + .all(wallet, now) as SignalRow[]; + return rows.map(rowToSignal); +} + +export async function cancelScheduledSignal( + id: string, + wallet: string, +): Promise { + const row = getSignalRow(id); + if (!row) throw new Error("Signal not found"); + if (row.author_wallet !== wallet) throw new Error("Not signal owner"); + if (row.status !== "scheduled") throw new Error("Signal is not scheduled"); + if ((row.scheduled_for ?? 0) <= Date.now()) + throw new Error("Signal has already published"); + getDb() + .prepare("UPDATE signals SET status = 'cancelled' WHERE id = ?") + .run(id); + return rowToSignal({ ...row, status: "cancelled" }); +} + +export async function voteOnPoll( + signalId: string, + optionId: string, + wallet: string, +): Promise { + const row = getSignalRow(signalId); + if (!row || row.type !== "poll" || !row.poll_json) + throw new Error("Poll not found"); + const poll = JSON.parse(row.poll_json) as SignalPoll; + if (poll.closes_at && poll.closes_at <= Date.now()) + throw new Error("Poll is closed"); + const selected = poll.options.find((option) => option.id === optionId); + if (!selected) throw new Error("Poll option not found"); + for (const option of poll.options) { + option.voters = option.voters.filter((voter) => voter !== wallet); } - return mutateJsonStore(filePath, (store) => { - store[signal.id] = signal - return signal - }) + selected.voters.push(wallet); + + getDb() + .prepare("UPDATE signals SET poll_json = ? WHERE id = ?") + .run(JSON.stringify(poll), signalId); + return rowToSignal({ ...row, poll_json: JSON.stringify(poll) }); } /** - * Toggle a wallet's upvote. Pass `expectedVersion` to make the toggle a - * compare-and-swap: a stale version throws VersionConflictError instead of - * silently discarding a concurrent writer's change. + * Bounded retries for the optimistic-concurrency loop below. A lost CAS means + * another writer committed between our read and our write; because the guard + * is version-based (not wall-clock), retries converge. 8 covers 50 concurrent + * writers with room to spare. + */ +const MAX_CAS_ATTEMPTS = 8; + +/** + * Toggle a wallet's upvote, guarded by the `version` column. + * + * `upvotes_json` is a JSON blob, so toggling is a read-modify-write. Two + * writers that both read the same row would otherwise both write back their + * own copy of the array and one upvote would vanish. The UPDATE is therefore + * guarded with `WHERE id = ? AND version = ?`: if another process committed in + * between, `changes` comes back 0 and we re-read and retry. + * + * Passing `expectedVersion` additionally makes this a client-driven + * compare-and-swap — a caller that composed its action against a stale view + * gets a VersionConflictError instead of clobbering the newer state. */ export async function upvoteSignal( id: string, wallet: string, expectedVersion?: number, ): Promise { - const filePath = serverDataJsonPath("signals") - return mutateJsonStore(filePath, (store) => { - const current = store[id] - if (!current) throw new Error("Signal not found") - const signal = normalizeSignal(current) - if (expectedVersion !== undefined && signal.version !== expectedVersion) { - throw new VersionConflictError(signal.version) + const db = getDb(); + const update = db.prepare( + `UPDATE signals + SET upvotes_json = ?, upvote_count = ?, version = ? + WHERE id = ? AND version = ?`, + ); + + for (let attempt = 0; attempt < MAX_CAS_ATTEMPTS; attempt++) { + const row = getSignalRow(id); + if (!row) throw new Error("Signal not found"); + if (expectedVersion !== undefined && row.version !== expectedVersion) { + throw new VersionConflictError(row.version); } - const idx = signal.upvotes.indexOf(wallet) + + const upvotes = JSON.parse(row.upvotes_json) as string[]; + const idx = upvotes.indexOf(wallet); if (idx === -1) { - signal.upvotes.push(wallet) + upvotes.push(wallet); } else { - signal.upvotes.splice(idx, 1) + upvotes.splice(idx, 1); } - signal.version += 1 - store[id] = signal - return signal - }) + const upvotesJson = JSON.stringify(upvotes); + const nextVersion = row.version + 1; + + const result = update.run( + upvotesJson, + upvotes.length, + nextVersion, + id, + row.version, + ); + if (result.changes === 1) { + return rowToSignal({ ...row, upvotes_json: upvotesJson, version: nextVersion }); + } + } + + // Every attempt lost the race. Surface the current state so the caller can + // retry against a fresh version rather than overwriting blindly. + const current = getSignalRow(id); + throw new VersionConflictError(current ? current.version : 0); } export async function getReplies(signal_id: string): Promise { - const store = await readJsonStore(serverDataJsonPath("signalReplies")) - return Object.values(store) - .filter((r) => r.signal_id === signal_id) - .map(normalizeReply) - .sort((a, b) => a.created_at - b.created_at) + const rows = getDb() + .prepare( + "SELECT * FROM signal_replies WHERE signal_id = ? ORDER BY created_at ASC", + ) + .all(signal_id) as ReplyRow[]; + return rows.map(rowToReply); } /** - * Append a reply. Pass `parentVersion` to reject a reply composed against a - * stale view of the parent signal. + * Append a reply. When `expectedParentVersion` is given the append is refused + * unless the parent signal is still at that version. + * + * The check runs inside the same `BEGIN IMMEDIATE` as the INSERT so the + * version cannot change between the test and the write. Callers that already + * compared the version before calling (the API does, to produce a 409 with the + * live version) pass it through so that pre-check stays advisory and this one + * is authoritative. */ export async function createReply( data: Omit, - parentVersion?: number, + expectedParentVersion?: number, ): Promise { - const signalsStore = await readJsonStore(serverDataJsonPath("signals")) - const parent = signalsStore[data.signal_id] - if (!parent) throw new Error("Signal not found") - const currentVersion = normalizeSignal(parent).version - if (parentVersion !== undefined && currentVersion !== parentVersion) { - throw new VersionConflictError(currentVersion) + const db = getDb(); + const reply: SignalReply = { + ...data, + id: nanoid(10), + created_at: Date.now(), + }; + + if (expectedParentVersion === undefined) { + db.prepare( + `INSERT INTO signal_replies + (id, signal_id, author_wallet, author_display, body, + upvotes_json, created_at, signature, signature_verified, media_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ).run( + reply.id, + reply.signal_id, + reply.author_wallet, + reply.author_display, + reply.body, + JSON.stringify(reply.upvotes ?? []), + reply.created_at, + reply.signature, + reply.signature_verified ? 1 : 0, + reply.media ? JSON.stringify(reply.media) : null, + ); + return reply; } - const filePath = serverDataJsonPath("signalReplies") - const reply: SignalReply = { ...data, id: nanoid(10), created_at: Date.now() } - return mutateJsonStore(filePath, (store) => { - store[reply.id] = reply - return reply - }) + db.exec("BEGIN IMMEDIATE"); + try { + const parent = getSignalRow(reply.signal_id); + if (!parent) throw new Error("Signal not found"); + if (parent.version !== expectedParentVersion) { + throw new VersionConflictError(parent.version); + } + db.prepare( + `INSERT INTO signal_replies + (id, signal_id, author_wallet, author_display, body, + upvotes_json, created_at, signature, signature_verified, media_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ).run( + reply.id, + reply.signal_id, + reply.author_wallet, + reply.author_display, + reply.body, + JSON.stringify(reply.upvotes ?? []), + reply.created_at, + reply.signature, + reply.signature_verified ? 1 : 0, + reply.media ? JSON.stringify(reply.media) : null, + ); + db.exec("COMMIT"); + } catch (error) { + db.exec("ROLLBACK"); + throw error; + } + return reply; } export async function getSignalChannelStats( @@ -937,7 +1175,7 @@ export async function editSignal( expectedVersion?: number, ): Promise<{ signal: Signal; version: SignalVersion }> { if (!isPhase82Enabled()) throw new SignalEditError("FLAG_DISABLED", "phase-82 disabled"); - if (expectedVersion !== undefined && (!Number.isInteger(expectedVersion) || expectedVersion < 0)) { + if (expectedVersion !== undefined && (!Number.isInteger(expectedVersion) || expectedVersion < 1)) { throw new SignalEditError("VALIDATION_FAILED", "A valid signal version is required"); } @@ -1054,7 +1292,7 @@ export function flag83RollbackNote(): string { return "Rollback phase-83: unset NEXT_PUBLIC_FEATURE_PHASE_83 / FEATURE_PHASE_83 or set to 0/false and restart. Reaction reads/writes become unavailable; existing signal_reactions rows remain on disk as inert history. No data migration to undo."; } -export const REACTION_EMOJI = ["👍", "❤️", "🔥", "😂", "😮", "😢"] as const; +export const REACTION_EMOJI = ["👍", "❤️", "🔥", "💡", "🎉", "🎯"] as const; export type ReactionEmoji = (typeof REACTION_EMOJI)[number]; export class SignalReactionError extends Error { diff --git a/lib/sqlite-db.ts b/lib/sqlite-db.ts index ce7262a..08d420a 100644 --- a/lib/sqlite-db.ts +++ b/lib/sqlite-db.ts @@ -63,7 +63,9 @@ CREATE INDEX IF NOT EXISTS idx_offers_status_wash_created CREATE TABLE IF NOT EXISTS signals ( id TEXT PRIMARY KEY, - version INTEGER NOT NULL DEFAULT 0, + -- Optimistic-concurrency token. Starts at 1 so that a fresh signal's version + -- is a valid ETag / If-Match / parent_version value (those all require >= 1). + version INTEGER NOT NULL DEFAULT 1, author_wallet TEXT NOT NULL, author_display TEXT NOT NULL, channel TEXT NOT NULL, @@ -236,11 +238,16 @@ export function getDb(): DatabaseSync { } try { db.exec( - "ALTER TABLE signals ADD COLUMN version INTEGER NOT NULL DEFAULT 0;", + "ALTER TABLE signals ADD COLUMN version INTEGER NOT NULL DEFAULT 1;", ); } catch { // Column already present — no-op. } + // The version column originally defaulted to 0, but every consumer of it + // (ETag, If-Match, parent_version) treats 0 as invalid, so a signal that had + // never been edited or upvoted could not be replied to. Normalize the + // sentinel to the same 1-based origin new rows get. + db.exec("UPDATE signals SET version = 1 WHERE version <= 0;"); return db; }