diff --git a/actions/publishToPublication.ts b/actions/publishToPublication.ts index c4fc26b9..9bf69af2 100644 --- a/actions/publishToPublication.ts +++ b/actions/publishToPublication.ts @@ -7,8 +7,11 @@ import { type PublishResult, } from "src/utils/publishLeaflet"; +// The rkey is server-chosen here: a caller-supplied one could target an +// existing record in the publication owner's repo. Only trusted server-side +// callers (the Ghost importer) pass one to publishLeaflet directly. export async function publishToPublication( - args: PublishLeafletArgs, + args: Omit, ): Promise { let identity = await getAuthIdentity(); if (!identity || !identity.atp_did) { @@ -21,5 +24,6 @@ export async function publishToPublication( }, }; } - return publishLeaflet({ ...args, actorDid: identity.atp_did }); + let { rkey: _ignored, ...safeArgs } = args as PublishLeafletArgs; + return publishLeaflet({ ...safeArgs, actorDid: identity.atp_did }); } diff --git a/app/(app)/(editor)/[leaflet_id]/actions/PublishButton.tsx b/app/(app)/(editor)/[leaflet_id]/actions/PublishButton.tsx index 36a27523..af55bafb 100644 --- a/app/(app)/(editor)/[leaflet_id]/actions/PublishButton.tsx +++ b/app/(app)/(editor)/[leaflet_id]/actions/PublishButton.tsx @@ -135,28 +135,40 @@ const UpdateButton = () => { labelOnMobile icon={} label={isLoading ? : "Update!"} + disabled={isLoading} onClick={async () => { - if (!pub) return; + if (!pub || isLoading) return; setIsLoading(true); - let result = await publishToPublication({ - root_entity: rootEntity, - publication_uri: pub.publications?.uri, - leaflet_id: permission_token.id, - title: currentTitle, - description: currentDescription, - tags: currentTags, - publishedAt: publishedAt?.toISOString(), - postPreferences, - }); - setIsLoading(false); - mutate(); + let result: Awaited>; + try { + result = await publishToPublication({ + root_entity: rootEntity, + publication_uri: pub.publications?.uri, + leaflet_id: permission_token.id, + title: currentTitle, + description: currentDescription, + tags: currentTags, + publishedAt: publishedAt?.toISOString(), + postPreferences, + }); + } catch (error) { + console.error(error); + toaster({ + content: "We couldn't publish this. Please try again!", + type: "error", + }); + return; + } finally { + setIsLoading(false); + mutate(); + } if (!result.success) { toaster({ content: isOAuthSessionError(result.error) ? ( ) : ( - "We couldn't publish this. Please try again!" + result.error.message ), type: "error", }); @@ -510,8 +522,10 @@ let useTitle = (entityID: string) => { // Only handle second block logic for linear documents, not canvas let isCanvas = canvasBlocks.length > 0; let secondBlock = !isCanvas ? blocks[1] : undefined; - let secondBlockTextValue = useEntity(secondBlock?.entityID || null, "block/text") - ?.data.value; + let secondBlockTextValue = useEntity( + secondBlock?.entityID || null, + "block/text", + )?.data.value; const secondBlockText = useMemo(() => { if (!secondBlockTextValue) return ""; let doc = new Y.Doc(); @@ -536,7 +550,13 @@ let useTitle = (entityID: string) => { etod.push(secondBlock.entityID); } return etod; - }, [firstBlockEntity, firstBlockType, secondBlockText, secondBlock, isCanvas]); + }, [ + firstBlockEntity, + firstBlockType, + secondBlockText, + secondBlock, + isCanvas, + ]); return { title: leafletTitle, entitiesToDelete }; }; diff --git a/app/(app)/(identity)/[leaflet_id]/publish/PublishPost.tsx b/app/(app)/(identity)/[leaflet_id]/publish/PublishPost.tsx index e0149d15..ccfceead 100644 --- a/app/(app)/(identity)/[leaflet_id]/publish/PublishPost.tsx +++ b/app/(app)/(identity)/[leaflet_id]/publish/PublishPost.tsx @@ -113,6 +113,7 @@ const PublishPostForm = ( let [oauthError, setOauthError] = useState< import("src/atproto-oauth").OAuthSessionError | null >(null); + let [publishError, setPublishError] = useState(null); let params = useParams(); let { rep } = useReplicache(); @@ -194,27 +195,41 @@ const PublishPostForm = ( if (isLoading) return; setIsLoading(true); setOauthError(null); - await rep?.push(); - let result = await publishToPublication({ - root_entity: props.root_entity, - publication_uri: props.publication_uri, - leaflet_id: props.leaflet_id, - title, - description, - tags: currentTags, - entitiesToDelete: props.entitiesToDelete, - publishedAt: localPublishedAt?.toISOString() || new Date().toISOString(), - postPreferences, - // Posting quietly forces every share channel off, mirroring the UI where - // checking "Post Quietly" unchecks all the other options. - sendEmail: shareState.email && !shareState.quiet, - showInDiscover: shareState.postToReaders && !shareState.quiet, - }); + setPublishError(null); + let result: Awaited>; + try { + await rep?.push(); + result = await publishToPublication({ + root_entity: props.root_entity, + publication_uri: props.publication_uri, + leaflet_id: props.leaflet_id, + title, + description, + tags: currentTags, + entitiesToDelete: props.entitiesToDelete, + publishedAt: + localPublishedAt?.toISOString() || new Date().toISOString(), + postPreferences, + // Posting quietly forces every share channel off, mirroring the UI where + // checking "Post Quietly" unchecks all the other options. + sendEmail: shareState.email && !shareState.quiet, + showInDiscover: shareState.postToReaders && !shareState.quiet, + }); + } catch (error) { + console.error(error); + setIsLoading(false); + setPublishError( + "Something went wrong while publishing. Please try again!", + ); + return; + } if (!result.success) { setIsLoading(false); if (isOAuthSessionError(result.error)) { setOauthError(result.error); + } else { + setPublishError(result.error.message); } return; } @@ -395,7 +410,7 @@ const PublishPostForm = ( 300 || nothingSelected} + disabled={charCount > 300 || nothingSelected || isLoading} > {isLoading ? ( @@ -410,6 +425,11 @@ const PublishPostForm = ( className="text-right text-sm text-accent-contrast" /> )} + {publishError && ( +
+ {publishError} +
+ )} )} diff --git a/src/utils/publishLeaflet.ts b/src/utils/publishLeaflet.ts index 6d262fce..5a271bc4 100644 --- a/src/utils/publishLeaflet.ts +++ b/src/utils/publishLeaflet.ts @@ -47,10 +47,18 @@ import { getDocumentType, } from "src/utils/collectionHelpers"; import { inngest } from "app/api/inngest/client"; +import { withPublishLock } from "src/utils/publishLock"; + +export type PublishError = + | OAuthSessionError + // Another publish of the same leaflet is still running (double submit, a + // second tab); the caller should wait for that one rather than retry. + | { type: "publish_in_progress"; message: string } + | { type: "publish_failed"; message: string }; export type PublishResult = | { success: true; rkey: string; record: SiteStandardDocument.Record } - | { success: false; error: OAuthSessionError }; + | { success: false; error: PublishError }; export type PublishLeafletArgs = { root_entity: string; @@ -80,7 +88,22 @@ export type PublishLeafletArgs = { // the trust boundary's inside: callers must have already established that the // actor may publish this leaflet (the server action checks the signed-in user; // admin tooling acts as the publication owner). -export async function publishLeaflet({ +export async function publishLeaflet( + args: PublishLeafletArgs & { actorDid: string }, +): Promise { + let result = await withPublishLock(args.leaflet_id, () => publish(args)); + if (!result.acquired) + return { + success: false, + error: { + type: "publish_in_progress", + message: "This post is already being published. Hang on a moment!", + }, + }; + return result.value; +} + +async function publish({ actorDid, root_entity, publication_uri, @@ -385,16 +408,20 @@ export async function publishLeaflet({ validate: false, //TODO publish the lexicon so we can validate! }); - // Optimistically create database entries - await supabaseServerClient.from("documents").upsert({ - uri: result.uri, - data: record as unknown as Json, - indexed: true, - }); - - if (publication_uri) { - // Publishing to a publication - update both tables - await Promise.all([ + // Link the record to the draft. A link write that fails silently leaves a + // published post the draft doesn't know about, and the next publish would + // mint a second one — so every write is checked, and a first publish whose + // link failed removes the record again so nothing is left behind. + let linkError = ( + await supabaseServerClient.from("documents").upsert({ + uri: result.uri, + data: record as unknown as Json, + indexed: true, + }) + ).error; + + if (!linkError && publication_uri) { + let results = await Promise.all([ supabaseServerClient.from("documents_in_publications").upsert({ publication: publication_uri, document: result.uri, @@ -409,16 +436,40 @@ export async function publishLeaflet({ tags: resolvedTags ?? [], }), ]); - } else { - // Publishing standalone - update leaflets_to_documents - await supabaseServerClient.from("leaflets_to_documents").upsert({ - leaflet: leaflet_id, - document: result.uri, - title: title || "", - description: description || "", - tags: resolvedTags ?? [], - }); + linkError = results.find((r) => r.error)?.error ?? null; + } else if (!linkError) { + linkError = ( + await supabaseServerClient.from("leaflets_to_documents").upsert({ + leaflet: leaflet_id, + document: result.uri, + title: title || "", + description: description || "", + tags: resolvedTags ?? [], + }) + ).error; + } + + if (linkError) { + console.error("Failed to link published document to draft", linkError); + if (!existingDocUri) { + await agent.com.atproto.repo + .deleteRecord({ + repo: credentialSession.did!, + collection: recordForPDS.$type, + rkey, + }) + .catch(console.error); + } + return { + success: false, + error: { + type: "publish_failed", + message: "We couldn't save this post. Please try again!", + }, + }; + } + if (!publication_uri) { // Heuristic: Remove title entities if this is the first time publishing standalone // (when entitiesToDelete is provided and there's no existing document) if (entitiesToDelete && entitiesToDelete.length > 0 && !existingDocUri) { diff --git a/src/utils/publishLock.test.ts b/src/utils/publishLock.test.ts new file mode 100644 index 00000000..b3e3da6b --- /dev/null +++ b/src/utils/publishLock.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it } from "vitest"; +import { withPublishLock } from "./publishLock"; + +function deferred() { + let resolve!: (v: T) => void; + let reject!: (e: unknown) => void; + let promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +describe("withPublishLock", () => { + it("refuses an overlapping publish of the same leaflet", async () => { + let first = deferred(); + let firstRun = withPublishLock("leaflet-a", () => first.promise); + let second = await withPublishLock("leaflet-a", async () => "second"); + expect(second).toEqual({ acquired: false }); + + first.resolve("first"); + expect(await firstRun).toEqual({ acquired: true, value: "first" }); + }); + + it("lets different leaflets publish concurrently", async () => { + let first = deferred(); + let firstRun = withPublishLock("leaflet-a", () => first.promise); + let other = await withPublishLock("leaflet-b", async () => "other"); + expect(other).toEqual({ acquired: true, value: "other" }); + first.resolve("first"); + await firstRun; + }); + + it("releases the lock once the publish settles, even on failure", async () => { + await expect( + withPublishLock("leaflet-a", async () => { + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + let retry = await withPublishLock("leaflet-a", async () => "retry"); + expect(retry).toEqual({ acquired: true, value: "retry" }); + }); +}); diff --git a/src/utils/publishLock.ts b/src/utils/publishLock.ts new file mode 100644 index 00000000..16f54d3b --- /dev/null +++ b/src/utils/publishLock.ts @@ -0,0 +1,78 @@ +import Client from "ioredis"; +import Redlock from "redlock"; + +// Publishing reads "does this draft already have a document?" and then writes +// one, so two overlapping publishes of the same draft each mint a fresh rkey +// and one of the resulting posts is left unlinked from the draft. Publishes +// are serialized per leaflet: an in-process set catches overlap within one +// server instance and, in production, a Redis lock catches overlap across +// instances. A busy caller is refused rather than queued — a queued duplicate +// would run as an "update" and hand its caller a second success (and a second +// Bluesky post). +const LOCK_TTL_MS = 5 * 60_000; +const inFlight = new Set(); + +const globalForPublishLock = globalThis as unknown as { + __publishRedlock?: Redlock | null; +}; + +function getRedlock(): Redlock | null { + if (globalForPublishLock.__publishRedlock === undefined) { + globalForPublishLock.__publishRedlock = + process.env.NODE_ENV === "production" && process.env.REDIS_URL + ? new Redlock([new Client(process.env.REDIS_URL)], { retryCount: 0 }) + : null; + } + return globalForPublishLock.__publishRedlock; +} + +// Redlock reports a failed acquire as an ExecutionError whose per-attempt +// votes carry the reason; only a ResourceLockedError vote means someone else +// holds the lock (anything else is Redis being unreachable, which should +// surface). The classes aren't reachable through the repo's redlock type shim, +// so they're matched by name. +type ExecutionStats = { votesAgainst: Map }; +async function isHeldElsewhere(error: unknown): Promise { + if (!(error instanceof Error) || error.name !== "ExecutionError") + return false; + let attempts = ( + error as { attempts?: ReadonlyArray> } + ).attempts; + let stats = await Promise.all(attempts ?? []); + return stats.some((attempt) => + [...attempt.votesAgainst.values()].some( + (vote) => vote.name === "ResourceLockedError", + ), + ); +} + +export type PublishLockResult = + | { acquired: true; value: T } + | { acquired: false }; + +export async function withPublishLock( + leaflet_id: string, + fn: () => Promise, +): Promise> { + if (inFlight.has(leaflet_id)) return { acquired: false }; + inFlight.add(leaflet_id); + try { + let redlock = getRedlock(); + let lock = null; + if (redlock) { + try { + lock = await redlock.acquire([`publish:${leaflet_id}`], LOCK_TTL_MS); + } catch (error) { + if (await isHeldElsewhere(error)) return { acquired: false }; + throw error; + } + } + try { + return { acquired: true, value: await fn() }; + } finally { + if (lock) await lock.release().catch(console.error); + } + } finally { + inFlight.delete(leaflet_id); + } +}