app/lib/publications/save.server.ts

Functions

Imports

It imports app/db/cv.ts, app/db/publications.ts, app/lib/cache-purge.server.ts, app/lib/carrel/errors.server.ts, app/lib/content/chunks.mjs, app/lib/cv/save.server.ts, app/lib/editor/converge.mjs, app/lib/editor/github.server.ts, app/lib/editor/publish-policy.mjs, app/lib/editor/write-path.server.ts, app/lib/error-message.mjs, app/lib/publications/cited-by.mjs, app/lib/publications/compile.mjs, app/lib/publications/parse.mjs, app/lib/publications/search-inputs.mjs, app/lib/publications/validate.mjs, app/lib/search/ask.server.ts, app/lib/search/records.mjs. Imported by app/lib/carrel/publication-handler.server.ts, app/lib/operator/api.server.ts, app/lib/operator/sync-tools.server.ts.

Source

1// A publication's write path, run for the Carrel adapter (app/lib/carrel/publication-handler.server.ts):
2// compile the file the way CI does (compile.mjs), refuse it with the validator's own messages if it fails,
3// commit it unless it is unchanged (commitUnlessUnchanged), then write its D1 row, its search record and
4// its cache purge, so the page changes on the next request with no build or deploy. The commit is never
5// reverted when a derived write fails: the error says the commit landed and how to repair it.
6
7import citedByArtifact from "../../../data/publications.cited-by.json";
8
9import { listCvCitedDois } from "~/db/cv";
10import { getPublicationRow, listPublicationIdentities } from "~/db/publications";
11import { textChunks } from "~/lib/content/chunks.mjs";
12import { ContentInvalid } from "~/lib/carrel/errors.server";
13import { combinePurges, purgeCv, purgePublications, type PurgeOutcome } from "~/lib/cache-purge.server";
14import { refreshCvSearch } from "~/lib/cv/save.server";
15import { convergeWithRetry } from "~/lib/editor/converge.mjs";
16import { blobGuard, commitFiles, publicFileEntry, readFile } from "~/lib/editor/github.server";
17import { decideFileWrite, type Actor } from "~/lib/editor/publish-policy.mjs";
18import { commitUnlessUnchanged, UNCHANGED_NOTE } from "~/lib/editor/write-path.server";
19import { errorMessage } from "~/lib/error-message.mjs";
20import { recordsForPapers } from "~/lib/search/records.mjs";
21import { askAvailable, syncAskPaper } from "~/lib/search/ask.server";
22
23import { citationSeedsFrom } from "./cited-by.mjs";
24import { compilePublication } from "./compile.mjs";
25import { parsePublication, publicationPath } from "./parse.mjs";
26import { paperSearchUid } from "./search-inputs.mjs";
27import type { OtherPublication, PublicationHost } from "./validate.mjs";
28
29type PublicationEnv = Env & { GITHUB_TOKEN?: string };
30
31/** A file the validator refused: carries every message, so the caller can fix its own edit. */
32class PublicationInvalid extends ContentInvalid {
33  constructor(slug: string, errors: string[]) {
34    super(`The publication "${slug}" was not saved: it fails ${errors.length} check(s). Nothing was committed.`, errors);
35    this.name = "PublicationInvalid";
36  }
37}
38
39/** The repository's PDFs, asked of GitHub: existence and size, never the bytes. */
40function githubHost(env: PublicationEnv): PublicationHost {
41  return { pdf: (sitePath) => publicFileEntry(env, sitePath) };
42}
43
44/** Every other publication's identity, from D1: CI re-checks the whole corpus on the commit this makes. */
45async function otherPublications(env: PublicationEnv, slug: string): Promise<OtherPublication[]> {
46  const rows = await listPublicationIdentities(env);
47  return rows
48    .filter((row) => row.slug !== slug)
49    .map((row) => {
50      const record = JSON.parse(row.record) as { id: string; doi: string | null };
51      return { slug: row.slug, doi: record.doi, id: record.id };
52    });
53}
54
55async function compile(env: PublicationEnv, slug: string, raw: string) {
56  return compilePublication({
57    slug,
58    raw,
59    host: githubHost(env),
60    others: await otherPublications(env, slug),
61    citedByArtifact,
62  });
63}
64
65type Compiled = Extract<Awaited<ReturnType<typeof compile>>, { ok: true }>;
66
67/** A compile for a read or a preview: the same door, so what a reader sees is what a save would write. */
68export async function compilePublicationFor(env: PublicationEnv, slug: string, raw: string) {
69  return compile(env, slug, raw);
70}
71
72/**
73 * The D1 row, its search record and the FTS rebuild, in one batch so they move together. D1 refuses a
74 * statement over 100 KB and a twin with the PDF's full text passes it, so the twin goes in as chunks (the
75 * way sync:content writes it): the row is inserted with the first chunk and a blank blob sha, the rest are
76 * appended, and the real sha lands last, so nothing reads a truncated twin as current. The batch is one
77 * transaction, so a failure leaves the previous row.
78 */
79export async function writePublicationRow(env: PublicationEnv, compiled: Compiled) {
80  const db = env.DB;
81  const r = compiled.record;
82  const uid = compiled.searchInput.uid;
83  const records = compiled.draft ? [] : recordsForPapers([compiled.searchInput]);
84  const [firstChunk = "", ...laterChunks] = textChunks(compiled.twin);
85  await db.batch([
86    db
87      .prepare(
88        `INSERT INTO publications (slug, doi_key, status, stage, type, title, year, selected, record, csl,
89           markdown, source_path, source_blob_sha, synced_at)
90         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, unixepoch())
91         ON CONFLICT(slug) DO UPDATE SET doi_key = excluded.doi_key, status = excluded.status, stage = excluded.stage,
92           type = excluded.type, title = excluded.title, year = excluded.year, selected = excluded.selected,
93           record = excluded.record, csl = excluded.csl, markdown = excluded.markdown,
94           source_path = excluded.source_path, source_blob_sha = excluded.source_blob_sha,
95           synced_at = excluded.synced_at`,
96      )
97      .bind(
98        r.slug,
99        r.doi ? r.doi.trim().toLowerCase() : null,
100        compiled.draft ? "draft" : "published",
101        r.stage,
102        r.type,
103        r.title,
104        r.year,
105        r.selected ? 1 : 0,
106        JSON.stringify(r),
107        compiled.csl ? JSON.stringify(compiled.csl) : null,
108        firstChunk,
109        compiled.sourcePath,
110        "",
111      ),
112    ...laterChunks.map((chunk) =>
113      db.prepare(`UPDATE publications SET markdown = markdown || ?1 WHERE slug = ?2`).bind(chunk, r.slug),
114    ),
115    db.prepare(`UPDATE publications SET source_blob_sha = ?1 WHERE slug = ?2`).bind(compiled.sourceBlobSha, r.slug),
116    db.prepare(`DELETE FROM search_docs WHERE doc_uid = ?1`).bind(uid),
117    ...records.map((s) =>
118      db
119        .prepare(
120          `INSERT INTO search_docs (uid, url, type, title, body, tags, doc_tags, doc_uid,
121             doc_title, doc_url, anchor, ordinal, status, publish_at)
122           VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)`,
123        )
124        .bind(s.uid, s.url, s.type, s.title, s.body, s.tags, s.docTags, s.docUid, s.docTitle, s.docUrl, s.anchor, s.ordinal, s.status, null),
125    ),
126    db.prepare(`INSERT INTO search_identity (search_identity) VALUES ('rebuild')`),
127    db.prepare(`INSERT INTO search_prose (search_prose) VALUES ('rebuild')`),
128  ]);
129}
130
131/**
132 * The row of a paper whose file is gone, its search record, and its citation row when no remaining paper
133 * carries that DOI. A paper that still has a file is never passed here.
134 */
135export async function deletePublicationRow(env: PublicationEnv, row: { slug: string; doi_key: string | null }) {
136  await env.DB.batch([
137    env.DB.prepare(`DELETE FROM publications WHERE slug = ?1`).bind(row.slug),
138    env.DB.prepare(`DELETE FROM search_docs WHERE doc_uid = ?1`).bind(paperSearchUid(row.slug)),
139    env.DB.prepare(
140      `DELETE FROM publication_citations WHERE doi = ?1 AND NOT EXISTS (SELECT 1 FROM publications WHERE doi_key = ?1)`,
141    ).bind(row.doi_key),
142    env.DB.prepare(`INSERT INTO search_identity (search_identity) VALUES ('rebuild')`),
143    env.DB.prepare(`INSERT INTO search_prose (search_prose) VALUES ('rebuild')`),
144  ]);
145}
146
147/**
148 * Seeds the committed OpenAlex snapshot's citation row for each DOI that has none. A row that exists is
149 * never touched: its count came from a refresh, which is newer than any snapshot.
150 */
151export async function seedMissingCitations(env: PublicationEnv, dois: Array<string | null>) {
152  const wanted = new Set(dois.filter((d): d is string => !!d).map((d) => d.trim().toLowerCase()));
153  const seeds = citationSeedsFrom(citedByArtifact, "data/publications.cited-by.json").filter((s) => wanted.has(s.doi));
154  if (seeds.length === 0) return;
155  await env.DB.batch(
156    seeds.map((s) =>
157      env.DB
158        .prepare(
159          `INSERT INTO publication_citations (doi, count, url, fetched_at) VALUES (?1, ?2, ?3, ?4)
160           ON CONFLICT(doi) DO NOTHING`,
161        )
162        .bind(s.doi, s.count, s.url, s.fetchedAt),
163    ),
164  );
165}
166
167/** True when D1's row for the paper was compiled from exactly this file (the git blob sha matches). */
168async function rowIsCurrent(env: PublicationEnv, slug: string, blobSha: string) {
169  const row = await env.DB.prepare("SELECT source_blob_sha FROM publications WHERE slug = ?1")
170    .bind(slug)
171    .first<{ source_blob_sha: string | null }>();
172  return row?.source_blob_sha === blobSha;
173}
174
175/** The Ask index must not fail a save: the commit has landed. A failure is returned, never dropped. */
176async function syncAsk(env: PublicationEnv, compiled: Compiled) {
177  if (!askAvailable(env)) return null;
178  try {
179    const result = await syncAskPaper(env, compiled.record.slug, compiled.draft ? null : compiled.twin);
180    return { ok: true as const, ...result };
181  } catch (error) {
182    console.error(JSON.stringify({ alert: "ask-sync-failed-after-publication-save", detail: errorMessage(error) }));
183    return {
184      ok: false as const,
185      message:
186        `The publication saved, but its Ask record did not update: ${errorMessage(error)}. ` +
187        `The next ship's Ask sync converges it from the repository.`,
188    };
189  }
190}
191
192/**
193 * What the CV (content/cv/publications.md) makes of this save. The CV cites a paper by DOI and is joined to the
194 * publications at read time, so a paper that stops being published, or whose DOI changes, would leave the CV
195 * citing a paper no record answers for: that is refused, naming the repair. A save of a paper the CV cites
196 * changes the CV, so `cited` tells the write to refresh the CV's search records and purge its cache tag.
197 */
198async function cvCitation(env: PublicationEnv, slug: string, compiled: Compiled) {
199  const cited = await listCvCitedDois(env);
200  const stored = cited.size === 0 ? null : await getPublicationRow(env, slug);
201  const storedKey = stored ? ((JSON.parse(stored.record) as { doi: string | null }).doi?.trim().toLowerCase() ?? null) : null;
202  if (!storedKey || !cited.has(storedKey)) return { cited: false, refusal: null };
203  const newKey = compiled.record.doi?.trim().toLowerCase() ?? null;
204  const refusal =
205    compiled.draft || newKey !== storedKey
206      ? `The CV cites this paper by DOI ${storedKey} (content/cv/publications.md), so it must stay published with that DOI. ` +
207        "Remove the reference from the CV first, or keep the paper published."
208      : null;
209  return { cited: true, refusal };
210}
211
212/** Whether a stored file is a draft, read from the file itself, the only authority. */
213export function fileIsDraft(raw: string) {
214  return parsePublication({ file: "", raw }).data.draft === true;
215}
216
217export type SavedPublication = {
218  slug: string;
219  commitSha: string;
220  /** The git blob sha of the committed file: Carrel's version of the item. */
221  sourceBlobSha: string;
222  unchanged: boolean;
223  created: boolean;
224  draft: boolean;
225  purged: PurgeOutcome;
226  askSync: Awaited<ReturnType<typeof syncAsk>>;
227  note?: string;
228};
229
230/**
231 * validate -> policy -> commitUnlessUnchanged -> blob verify -> D1 write (convergeWithRetry) -> cache
232 * purge by tag, as saveProcedure does it.
233 */
234export async function savePublication(
235  env: PublicationEnv,
236  options: { slug: string; raw: string; expectedHeadSha?: string | null;
237    /** The blob sha of this file as Carrel loaded it; undefined means no check. */
238    expectedBlobSha?: string | null; isNew: boolean; actor: Actor },
239): Promise<SavedPublication> {
240  const { slug, raw, actor } = options;
241  const path = publicationPath(slug);
242  const existing = await readFile(env, path);
243  if (options.isNew && existing) throw new PublicationInvalid(slug, [`a publication named ${slug} already exists`]);
244  if (!options.isNew && !existing) throw new PublicationInvalid(slug, [`no publication named ${slug} exists; pass isNew to create it`]);
245
246  const compiled = await compile(env, slug, raw);
247  if (!compiled.ok) throw new PublicationInvalid(slug, compiled.errors);
248  decideFileWrite({
249    actor,
250    noun: "publication",
251    incomingDraft: compiled.draft,
252    priorRaw: existing?.content ?? null,
253    isDraft: fileIsDraft,
254  });
255
256  const cv = await cvCitation(env, slug, compiled);
257  if (cv.refusal) throw new PublicationInvalid(slug, [cv.refusal]);
258
259  const tag = actor.kind === "operator" ? ` [operator:${actor.id}]` : actor.kind === "carrel" ? ` [carrel:${actor.changeId}]` : "";
260  // After the gate, so a file that fails validation is still refused even when it is unchanged.
261  const written = await commitUnlessUnchanged(env, {
262    existing,
263    raw,
264    rowIsCurrent: () => rowIsCurrent(env, slug, compiled.sourceBlobSha),
265    commit: () =>
266      commitFiles(env, {
267        expectedHeadSha: options.expectedHeadSha,
268        expectedBlobs: blobGuard(path, options.expectedBlobSha),
269        message: `${options.isNew ? "Add" : "Update"} publication: ${compiled.record.title}${tag}`,
270        changes: [{ path, content: raw }],
271      }),
272  });
273  if (written.action === "noop") {
274    return {
275      slug,
276      commitSha: written.commitSha,
277      sourceBlobSha: compiled.sourceBlobSha,
278      unchanged: true,
279      created: false,
280      draft: compiled.draft,
281      purged: null,
282      askSync: null,
283      note: UNCHANGED_NOTE,
284    };
285  }
286  const { commitSha } = written;
287  const blobSha = written.blobShas[path];
288  if (blobSha && blobSha !== compiled.sourceBlobSha) {
289    throw new Error(
290      `The publication "${slug}" WAS committed as ${commitSha}, but the committed bytes (${blobSha}) are not ` +
291        `the bytes compiled (${compiled.sourceBlobSha}), so its page was not updated. Run the content sync to ` +
292        `rebuild the row from the repository.`,
293    );
294  }
295
296  let purged: PurgeOutcome = false;
297  await convergeWithRetry({
298    write: async () => {
299      await writePublicationRow(env, compiled);
300      purged = await purgePublications(`save_publication ${slug}`);
301      // The CV is joined to this paper at read time, so its page reads the new row already; its search
302      // records and its cache are what a save has to refresh.
303      if (cv.cited) {
304        await refreshCvSearch(env);
305        purged = combinePurges(purged, await purgeCv(`save_publication ${slug}`));
306      }
307    },
308    // Nothing kept in KV for a publication: the thrown error names the commit and the repair (the content
309    // sync), and the next ship's sync converges the row from the repository anyway.
310    recordDivergence: async () => undefined,
311    slug,
312    commitSha,
313  });
314
315  return {
316    slug,
317    commitSha,
318    sourceBlobSha: compiled.sourceBlobSha,
319    unchanged: written.action === "repair",
320    created: options.isNew,
321    draft: compiled.draft,
322    purged,
323    askSync: await syncAsk(env, compiled),
324  };
325}