From 51808f47e8dfb291bf66a528b9735cdcbe59d2e5 Mon Sep 17 00:00:00 2001 From: Lichen Dev Agent Date: Sat, 22 Aug 2026 22:59:30 +0000 Subject: [PATCH] Open wiki mode M3: firehose ingest - wiki records: persist openContributions flag - ingest wiki.lichen.contribution -> pending_contributions (queue, not live) - ingest wiki.lichen.contributionApproval -> status + optional contributor grant - drops: contribution on non-open wiki, from a direct editor, non-admin approval - firehose tests for flag, queueing, drops, approve/promote, reject --- src/firehose/handlers.ts | 165 ++++++++++++++- src/server/db/queries/index.ts | 4 + src/server/db/queries/pending-contribution.ts | 65 ++++++ src/server/db/queries/wiki.ts | 35 +++- tests/firehose/handlers.test.ts | 194 ++++++++++++++++++ 5 files changed, 456 insertions(+), 7 deletions(-) diff --git a/src/firehose/handlers.ts b/src/firehose/handlers.ts index bf45c5e..0878fe5 100644 --- a/src/firehose/handlers.ts +++ b/src/firehose/handlers.ts @@ -1,5 +1,5 @@ import { isDid, isRecordKey } from "@atcute/lexicons/syntax"; -import { canEdit, getAccessLevel } from "../lib/access.ts"; +import { canEdit, canManage, getAccessLevel } from "../lib/access.ts"; import { COLLECTIONS, normalizeRole } from "../lib/collections.ts"; import { isValidLanguageTag } from "../lib/languages.ts"; import { LIMITS } from "../lib/limits.ts"; @@ -16,17 +16,21 @@ import { deleteBookmarkByUri, deleteMembershipByUri, deleteNoteByAtUri, + deletePendingContribution, deleteRequestByUri, deleteWikiByAtUri, getMemberRole, getNoteByAtUri, getNoteBySlug, + getPendingContributionByAtUri, getWikiByAtUri, isDidBanned, + setContributionDecision, setWikiTheme, upsertBookmark, upsertMembership, upsertNote, + upsertPendingContribution, upsertRequest, upsertWiki, } from "../server/db/queries/index.ts"; @@ -40,6 +44,7 @@ interface WikiRecord { language?: string; description?: string; theme?: string; + openContributions?: boolean; } interface NoteRecord { @@ -83,6 +88,25 @@ interface CommunityBookmarkRecord { createdAt: string; } +// Open-wiki contribution envelope (author = submitter). `note` and `noteRevision` +// are com.atproto.repo.strongRef objects; we only need their uris to materialise +// the review queue. +interface ContributionRecord { + wikiRef: string; + kind: "create-note" | "edit-note"; + note: { uri: string }; + noteRevision: { uri: string }; + createdAt: string; +} + +// Owner/admin decision on a pending contribution. +interface ContributionApprovalRecord { + contributionRef: string; + decision: "approve" | "reject"; + promote?: boolean; + createdAt: string; +} + // Remote records honour neither our limits nor our field shapes. The rule below: // reject only what we cannot represent, coerce the rest — handlers resolve by // reference, so dropping a wiki orphans every note under it. @@ -258,6 +282,33 @@ function isCommunityBookmarkRecord(r: Rec): r is Rec & CommunityBookmarkRecord { return hasNonEmptyString(r, "subject") && hasNonEmptyString(r, "createdAt"); } +// Returns the uri member of a strongRef-typed value, or null if it isn't one. +function strongRefUri(v: unknown): string | null { + if (typeof v !== "object" || v === null) return null; + const o = v as Rec; + return typeof o["uri"] === "string" && o["uri"] !== "" ? o["uri"] : null; +} + +function isContributionRecord(r: Rec): r is Rec & ContributionRecord { + return ( + hasNonEmptyString(r, "wikiRef") && + (r["kind"] === "create-note" || r["kind"] === "edit-note") && + strongRefUri(r["note"]) !== null && + strongRefUri(r["noteRevision"]) !== null && + hasNonEmptyString(r, "createdAt") + ); +} + +function isContributionApprovalRecord( + r: Rec, +): r is Rec & ContributionApprovalRecord { + return ( + hasNonEmptyString(r, "contributionRef") && + (r["decision"] === "approve" || r["decision"] === "reject") && + hasNonEmptyString(r, "createdAt") + ); +} + // Built from a jetstream message in prod; tests invoke handleCommitEvent directly. export interface FirehoseCommit { did: string; @@ -351,6 +402,22 @@ export function handleCommitEvent(evt: FirehoseCommit): void { if (isCommunityBookmarkRecord(r)) handleCommunityBookmark(evt.did, atUri, r); break; + case COLLECTIONS.contribution: { + if (!isContributionRecord(r)) { + malformed(); + break; + } + handleContribution(evt.did, atUri, r); + break; + } + case COLLECTIONS.contributionApproval: { + if (!isContributionApprovalRecord(r)) { + malformed(); + break; + } + handleContributionApproval(evt.did, atUri, r); + break; + } } } @@ -385,6 +452,7 @@ function handleWiki( record.createdAt, record.language ?? "en", record.description ?? "", + record.openContributions ? 1 : 0, ); // Per the lexicon, absent means "let the viewer choose" — reader mode. The @@ -523,6 +591,95 @@ function handleCommunityBookmark( upsertBookmark(did, record.subject, atUri, record.createdAt); } +function handleContribution( + did: string, + atUri: string, + record: ContributionRecord, +): void { + const wiki = getWikiByAtUri(record.wikiRef); + if (!wiki) { + logDrop(atUri, `contribution.wikiRef unknown: ${record.wikiRef}`); + return; + } + + // Only public, open wikis accept contributions, and a direct editor routes + // their work straight to live — an envelope from one is dropped. + if (wiki.visibility !== "public" || wiki.contributions_open !== 1) { + logDrop(atUri, `contribution targets a non-open wiki (${wiki.at_uri})`); + return; + } + const role = getMemberRole(wiki.at_uri, did); + if (canEdit(getAccessLevel(wiki, did, role))) { + logDrop(atUri, `${did} is a direct editor of ${wiki.at_uri}`); + return; + } + + const noteUri = strongRefUri(record.note); + const noteRevisionUri = strongRefUri(record.noteRevision); + if (noteUri === null || noteRevisionUri === null) { + logDrop(atUri, "contribution.note/noteRevision missing uri"); + return; + } + + upsertPendingContribution( + atUri, + wiki.at_uri, + wiki.slug, + noteUri, + noteRevisionUri, + record.kind, + did, + ); +} + +function handleContributionApproval( + did: string, + atUri: string, + record: ContributionApprovalRecord, +): void { + const pending = getPendingContributionByAtUri(record.contributionRef); + if (!pending) { + logDrop( + atUri, + `approval for unknown contribution ${record.contributionRef}`, + ); + return; + } + + const wiki = getWikiByAtUri(pending.wiki_at_uri); + if (!wiki) { + logDrop(atUri, `approval's wiki unknown: ${pending.wiki_at_uri}`); + return; + } + + // Only the wiki owner or an admin may decide contributions. + const role = getMemberRole(wiki.at_uri, did); + if (!canManage(getAccessLevel(wiki, did, role))) { + logDrop(atUri, `${did} may not decide contributions on ${wiki.at_uri}`); + return; + } + + if (record.decision === "reject") { + setContributionDecision(pending.contribution_at_uri, "rejected", did); + return; + } + + setContributionDecision(pending.contribution_at_uri, "approved", did); + if (record.promote) { + // Approve-and-make-contributor: grant the submitter a contributor + // membership. The approval record's own at-uri serves as the membership + // at-uri (synthetic, local); upsertMembership keys on (wiki, did). + upsertMembership( + wiki.at_uri, + wiki.slug, + pending.submitter_did, + "contributor", + atUri, + record.createdAt, + ); + } +} + function handleDelete(atUri: string, collection: string): void { switch (collection) { case COLLECTIONS.wiki: @@ -543,5 +700,11 @@ function handleDelete(atUri: string, collection: string): void { case COLLECTIONS.communityBookmark: deleteBookmarkByUri(atUri); break; + case COLLECTIONS.contribution: + deletePendingContribution(atUri); + break; + case COLLECTIONS.contributionApproval: + // Decisions are derived into pending_contributions; nothing to remove. + break; } } diff --git a/src/server/db/queries/index.ts b/src/server/db/queries/index.ts index 4771b51..d5533ef 100644 --- a/src/server/db/queries/index.ts +++ b/src/server/db/queries/index.ts @@ -57,8 +57,12 @@ export { export { type ContributionKind, type ContributionStatus, + deletePendingContribution, + getPendingContributionByAtUri, insertPendingContribution, type PendingContributionRow, + setContributionDecision, + upsertPendingContribution, } from "./pending-contribution.ts"; export { expireCachedProfile, diff --git a/src/server/db/queries/pending-contribution.ts b/src/server/db/queries/pending-contribution.ts index 1182f5c..521d26d 100644 --- a/src/server/db/queries/pending-contribution.ts +++ b/src/server/db/queries/pending-contribution.ts @@ -43,3 +43,68 @@ export function insertPendingContribution( ], ); } + +// Firehose path: a submission envelope may already have been recorded by the +// direct HTTP submit. Treat a duplicate as already-known (never overwrite). +export function upsertPendingContribution( + contributionAtUri: string, + wikiAtUri: string, + wikiSlug: string, + noteAtUri: string, + noteRevisionAtUri: string, + kind: ContributionKind, + submitterDid: string, +): void { + const db = getDb(); + db.run( + `INSERT INTO pending_contributions + (contribution_at_uri, wiki_at_uri, wiki_slug, note_at_uri, note_revision_at_uri, kind, submitter_did, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now')) + ON CONFLICT(contribution_at_uri) DO NOTHING`, + [ + contributionAtUri, + wikiAtUri, + wikiSlug, + noteAtUri, + noteRevisionAtUri, + kind, + submitterDid, + ], + ); +} + +export function getPendingContributionByAtUri( + contributionAtUri: string, +): PendingContributionRow | null { + const db = getDb(); + return ( + (db + .query( + "SELECT * FROM pending_contributions WHERE contribution_at_uri = ?", + ) + .get(contributionAtUri) as PendingContributionRow) ?? null + ); +} + +// Apply an owner/admin's decision on a pending contribution (approved/rejected), +// recording who decided. Promotion-to-live is handled separately from this. +export function setContributionDecision( + contributionAtUri: string, + status: "approved" | "rejected", + approvedByDid: string, +): void { + const db = getDb(); + db.run( + `UPDATE pending_contributions + SET status = ?, approved_by_did = ?, approved_at = datetime('now') + WHERE contribution_at_uri = ?`, + [status, approvedByDid, contributionAtUri], + ); +} + +export function deletePendingContribution(contributionAtUri: string): void { + const db = getDb(); + db.run("DELETE FROM pending_contributions WHERE contribution_at_uri = ?", [ + contributionAtUri, + ]); +} diff --git a/src/server/db/queries/wiki.ts b/src/server/db/queries/wiki.ts index bec5770..a8e9cca 100644 --- a/src/server/db/queries/wiki.ts +++ b/src/server/db/queries/wiki.ts @@ -208,12 +208,23 @@ export function insertWiki( createdAt: string, language = "en", description = "", + openContributions = 0, ): void { const db = getDb(); db.run( - `INSERT INTO wikis (slug, did, name, visibility, language, description, at_uri, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))`, - [slug, did, name, visibility, language, description, atUri, createdAt], + `INSERT INTO wikis (slug, did, name, visibility, language, description, contributions_open, at_uri, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))`, + [ + slug, + did, + name, + visibility, + language, + description, + openContributions, + atUri, + createdAt, + ], ); } @@ -226,18 +237,30 @@ export function upsertWiki( createdAt: string, language = "en", description = "", + openContributions = 0, ): void { const db = getDb(); db.run( - `INSERT INTO wikis (slug, did, name, visibility, language, description, at_uri, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) + `INSERT INTO wikis (slug, did, name, visibility, language, description, contributions_open, at_uri, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) ON CONFLICT(did, slug) DO UPDATE SET name = excluded.name, visibility = excluded.visibility, language = excluded.language, description = excluded.description, + contributions_open = excluded.contributions_open, updated_at = datetime('now')`, - [slug, did, name, visibility, language, description, atUri, createdAt], + [ + slug, + did, + name, + visibility, + language, + description, + openContributions, + atUri, + createdAt, + ], ); } diff --git a/tests/firehose/handlers.test.ts b/tests/firehose/handlers.test.ts index 2ddc2ac..aeb8495 100644 --- a/tests/firehose/handlers.test.ts +++ b/tests/firehose/handlers.test.ts @@ -9,6 +9,7 @@ import { getNoteBySlug, getWiki, isBookmarked, + type PendingContributionRow, upsertBookmark, } from "../../src/server/db/queries/index.ts"; @@ -50,6 +51,8 @@ const HANDLER_TEST_WIKIS = [ "theme-unknown", "theme-cleared", "cbk-wiki", + "open-wiki", + "closed-wiki", ]; function cleanupHandlerTestData() { @@ -936,3 +939,194 @@ describe("size guards", () => { expect(getCurrentNote(WIKI_AT_URI, "oversize-note")).toBeNull(); }); }); + +describe("open-contribution firehose ingest", () => { + const OPEN = `at://${ALICE_DID}/wiki.lichen.wiki/open-wiki`; + const CLOSED = `at://${ALICE_DID}/wiki.lichen.wiki/closed-wiki`; + const CAROL_DID = "did:plc:carol"; + const DAVE_DID = "did:plc:dave"; + + function createWiki(slug: string, open: boolean) { + handleCommitEvent( + makeCommitEvt({ + event: "create", + collection: "wiki.lichen.wiki", + rkey: slug, + did: ALICE_DID, + record: { + name: slug, + visibility: "public", + createdAt: "2026-01-01T00:00:00.000Z", + ...(open ? { openContributions: true } : {}), + }, + }), + ); + } + + function submitContribution( + did: string, + rkey: string, + wikiRef: string, + kind = "create-note", + ) { + handleCommitEvent( + makeCommitEvt({ + event: "create", + collection: "wiki.lichen.contribution", + rkey, + did, + record: { + wikiRef, + kind, + note: { + uri: `at://${did}/wiki.lichen.note/${rkey}n`, + cid: "bafyrei1", + }, + noteRevision: { + uri: `at://${did}/wiki.lichen.noteRevision/${rkey}r`, + cid: "bafyrei2", + }, + createdAt: "2026-01-02T00:00:00.000Z", + }, + }), + ); + } + + function pendingRow(atUri: string): PendingContributionRow | null { + return ( + (db + .query( + "SELECT * FROM pending_contributions WHERE contribution_at_uri = ?", + ) + .get(atUri) as PendingContributionRow) ?? null + ); + } + + beforeAll(() => { + createWiki("open-wiki", true); + createWiki("closed-wiki", false); + }); + + test("persists the openContributions flag from the firehose wiki record", () => { + expect(getWiki(ALICE_DID, "open-wiki")?.contributions_open).toBe(1); + expect(getWiki(ALICE_DID, "closed-wiki")?.contributions_open).toBe(0); + }); + + test("queues a non-contributor's contribution on an open wiki (not live)", () => { + submitContribution(BOB_DID, "bobs-contrib", OPEN); + + const row = pendingRow( + `at://${BOB_DID}/wiki.lichen.contribution/bobs-contrib`, + ); + expect(row).not.toBeNull(); + expect(row?.kind).toBe("create-note"); + expect(row?.submitter_did).toBe(BOB_DID); + expect(row?.status).toBe("pending"); + + // The proposed note is NOT made live. + const live = db + .query("SELECT * FROM notes WHERE wiki_at_uri = ? AND slug = ?") + .get(OPEN, "bobs-contrib"); + expect(live).toBeNull(); + }); + + test("drops a contribution targeting a non-open wiki", () => { + submitContribution(BOB_DID, "bobs-closed", CLOSED); + expect( + pendingRow(`at://${BOB_DID}/wiki.lichen.contribution/bobs-closed`), + ).toBeNull(); + }); + + test("drops a direct editor's contribution on an open wiki", () => { + // ALICE (owner) grants BOB contributor status. + handleCommitEvent( + makeCommitEvt({ + event: "create", + collection: "wiki.lichen.membership", + rkey: "membership-bob", + did: ALICE_DID, + record: { + memberDid: BOB_DID, + wikiRef: OPEN, + role: "contributor", + createdAt: "2026-01-02T00:00:00.000Z", + }, + }), + ); + submitContribution(BOB_DID, "bobs-as-contributor", OPEN); + expect( + pendingRow( + `at://${BOB_DID}/wiki.lichen.contribution/bobs-as-contributor`, + ), + ).toBeNull(); + }); + + test("owner can approve-and-promote a pending contribution", () => { + submitContribution(CAROL_DID, "carol-contrib", OPEN); + const contribAtUri = `at://${CAROL_DID}/wiki.lichen.contribution/carol-contrib`; + + handleCommitEvent( + makeCommitEvt({ + event: "create", + collection: "wiki.lichen.contributionApproval", + rkey: "approval-carol", + did: ALICE_DID, + record: { + contributionRef: contribAtUri, + decision: "approve", + promote: true, + createdAt: "2026-01-03T00:00:00.000Z", + }, + }), + ); + + expect(pendingRow(contribAtUri)?.status).toBe("approved"); + const member = db + .query("SELECT role FROM memberships WHERE wiki_at_uri = ? AND did = ?") + .get(OPEN, CAROL_DID) as { role: string } | null; + expect(member?.role).toBe("contributor"); + }); + + test("owner can reject a pending contribution", () => { + submitContribution(DAVE_DID, "dave-contrib", OPEN); + const contribAtUri = `at://${DAVE_DID}/wiki.lichen.contribution/dave-contrib`; + + handleCommitEvent( + makeCommitEvt({ + event: "create", + collection: "wiki.lichen.contributionApproval", + rkey: "reject-dave", + did: ALICE_DID, + record: { + contributionRef: contribAtUri, + decision: "reject", + createdAt: "2026-01-03T00:00:00.000Z", + }, + }), + ); + + expect(pendingRow(contribAtUri)?.status).toBe("rejected"); + }); + + test("a non-admin approval is dropped (decision unchanged)", () => { + submitContribution(DAVE_DID, "dave-contrib2", OPEN); + const contribAtUri = `at://${DAVE_DID}/wiki.lichen.contribution/dave-contrib2`; + + // BOB is a contributor (not admin) on the open wiki — may not decide. + handleCommitEvent( + makeCommitEvt({ + event: "create", + collection: "wiki.lichen.contributionApproval", + rkey: "bad-approval", + did: BOB_DID, + record: { + contributionRef: contribAtUri, + decision: "approve", + createdAt: "2026-01-03T00:00:00.000Z", + }, + }), + ); + + expect(pendingRow(contribAtUri)?.status).toBe("pending"); + }); +}); -- 2.51.2