diff --git a/actions/publishToPublication.ts b/actions/publishToPublication.ts index bad2144c..cf1b2a82 100644 --- a/actions/publishToPublication.ts +++ b/actions/publishToPublication.ts @@ -199,7 +199,9 @@ export async function publishToPublication({ } // Determine the collection to use - preserve existing schema if updating - const existingCollection = existingDocUri ? new AtUri(existingDocUri).collection : undefined; + const existingCollection = existingDocUri + ? new AtUri(existingDocUri).collection + : undefined; const documentType = getDocumentType(existingCollection); // Build the pages array (used by both formats) @@ -228,13 +230,14 @@ export async function publishToPublication({ if (documentType === "site.standard.document") { // site.standard.document format // For standalone docs, use HTTPS URL; for publication docs, use the publication AT-URI - const siteUri = publication_uri || `https://leaflet.pub/p/${credentialSession.did}`; + const siteUri = + publication_uri || `https://leaflet.pub/p/${credentialSession.did}`; record = { $type: "site.standard.document", title: title || "Untitled", site: siteUri, - path: rkey, + path: "/" + rkey, publishedAt: publishedAt || existingRecord.publishedAt || new Date().toISOString(), ...(description && { description }), -- 2.51.2 From 8dc7f9b8342f0e98a7babd60b246c45ce313a443 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 16:02:09 -0500 Subject: [PATCH 02/12] fix some lexicon validation stuff --- .../functions/migrate_user_to_standard.ts | 78 +++++++------------ lexicons/api/lexicons.ts | 4 +- .../api/types/site/standard/publication.ts | 2 +- lexicons/site/standard/publication.json | 4 +- lexicons/src/normalize.ts | 45 +++++++++-- 5 files changed, 75 insertions(+), 58 deletions(-) diff --git a/app/api/inngest/functions/migrate_user_to_standard.ts b/app/api/inngest/functions/migrate_user_to_standard.ts index 7c630e42..cf809b64 100644 --- a/app/api/inngest/functions/migrate_user_to_standard.ts +++ b/app/api/inngest/functions/migrate_user_to_standard.ts @@ -109,10 +109,11 @@ export const migrate_user_to_standard = inngest.createFunction( }) .filter((x) => x !== null); - // Run all PDS writes in parallel - const pubPdsResults = await Promise.all( - publicationsToMigrate.map(({ pub, rkey, newRecord }) => - step.run(`pds-write-publication-${pub.uri}`, async () => { + // Run PDS + DB writes together for each publication + const pubResults = await Promise.all( + publicationsToMigrate.map(({ pub, rkey, normalized, newRecord }) => + step.run(`migrate-publication-${pub.uri}`, async () => { + // PDS write const agent = await createAuthenticatedAgent(did); const putResult = await agent.com.atproto.repo.putRecord({ repo: did, @@ -121,16 +122,9 @@ export const migrate_user_to_standard = inngest.createFunction( record: newRecord, validate: false, }); - return { oldUri: pub.uri, newUri: putResult.data.uri }; - }), - ), - ); + const newUri = putResult.data.uri; - // Run all DB writes in parallel - const pubDbResults = await Promise.all( - publicationsToMigrate.map(({ pub, normalized, newRecord }, index) => { - const newUri = pubPdsResults[index].newUri; - return step.run(`db-write-publication-${pub.uri}`, async () => { + // DB write const { error: dbError } = await supabaseServerClient .from("publications") .upsert({ @@ -149,12 +143,12 @@ export const migrate_user_to_standard = inngest.createFunction( }; } return { success: true as const, oldUri: pub.uri, newUri }; - }); - }), + }), + ), ); // Process results - for (const result of pubDbResults) { + for (const result of pubResults) { if (result.success) { publicationUriMap[result.oldUri] = result.newUri; stats.publicationsMigrated++; @@ -239,7 +233,7 @@ export const migrate_user_to_standard = inngest.createFunction( $type: "site.standard.document", title: normalized.title || "Untitled", site: siteValue, - path: rkey, + path: "/" + rkey, publishedAt: normalized.publishedAt || new Date().toISOString(), description: normalized.description, content: normalized.content, @@ -252,10 +246,11 @@ export const migrate_user_to_standard = inngest.createFunction( }) .filter((x) => x !== null); - // Run all PDS writes in parallel - const docPdsResults = await Promise.all( - documentsToMigrate.map(({ doc, rkey, newRecord }) => - step.run(`pds-write-document-${doc.uri}`, async () => { + // Run PDS + DB writes together for each document + const docResults = await Promise.all( + documentsToMigrate.map(({ doc, rkey, newRecord, oldPubUri }) => + step.run(`migrate-document-${doc.uri}`, async () => { + // PDS write const agent = await createAuthenticatedAgent(did); const putResult = await agent.com.atproto.repo.putRecord({ repo: did, @@ -264,16 +259,9 @@ export const migrate_user_to_standard = inngest.createFunction( record: newRecord, validate: false, }); - return { oldUri: doc.uri, newUri: putResult.data.uri }; - }), - ), - ); + const newUri = putResult.data.uri; - // Run all DB writes in parallel - const docDbResults = await Promise.all( - documentsToMigrate.map(({ doc, newRecord, oldPubUri }, index) => { - const newUri = docPdsResults[index].newUri; - return step.run(`db-write-document-${doc.uri}`, async () => { + // DB write const { error: dbError } = await supabaseServerClient .from("documents") .upsert({ @@ -302,12 +290,12 @@ export const migrate_user_to_standard = inngest.createFunction( } return { success: true as const, oldUri: doc.uri, newUri }; - }); - }), + }), + ), ); // Process results - for (const result of docDbResults) { + for (const result of docResults) { if (result.success) { documentUriMap[result.oldUri] = result.newUri; stats.documentsMigrated++; @@ -428,10 +416,11 @@ export const migrate_user_to_standard = inngest.createFunction( }) .filter((x) => x !== null); - // Run all PDS writes in parallel - const subPdsResults = await Promise.all( + // Run PDS + DB writes together for each subscription + const subResults = await Promise.all( subscriptionsToMigrate.map(({ sub, rkey, newRecord }) => - step.run(`pds-write-subscription-${sub.uri}`, async () => { + step.run(`migrate-subscription-${sub.uri}`, async () => { + // PDS write const agent = await createAuthenticatedAgent(did); const putResult = await agent.com.atproto.repo.putRecord({ repo: did, @@ -440,16 +429,9 @@ export const migrate_user_to_standard = inngest.createFunction( record: newRecord, validate: false, }); - return { oldUri: sub.uri, newUri: putResult.data.uri }; - }), - ), - ); + const newUri = putResult.data.uri; - // Run all DB writes in parallel - const subDbResults = await Promise.all( - subscriptionsToMigrate.map(({ sub, newRecord }, index) => { - const newUri = subPdsResults[index].newUri; - return step.run(`db-write-subscription-${sub.uri}`, async () => { + // DB write const { error: dbError } = await supabaseServerClient .from("publication_subscriptions") .update({ @@ -467,12 +449,12 @@ export const migrate_user_to_standard = inngest.createFunction( }; } return { success: true as const, oldUri: sub.uri, newUri }; - }); - }), + }), + ), ); // Process results - for (const result of subDbResults) { + for (const result of subResults) { if (result.success) { userSubscriptionUriMap[result.oldUri] = result.newUri; stats.userSubscriptionsMigrated++; diff --git a/lexicons/api/lexicons.ts b/lexicons/api/lexicons.ts index 409eb985..5ef436e0 100644 --- a/lexicons/api/lexicons.ts +++ b/lexicons/api/lexicons.ts @@ -2215,8 +2215,8 @@ export const schemaDict = { type: 'ref', }, theme: { - type: 'ref', - ref: 'lex:pub.leaflet.publication#theme', + type: 'union', + refs: ['lex:pub.leaflet.publication#theme'], }, description: { maxGraphemes: 300, diff --git a/lexicons/api/types/site/standard/publication.ts b/lexicons/api/types/site/standard/publication.ts index 88f3b671..2daf12b6 100644 --- a/lexicons/api/types/site/standard/publication.ts +++ b/lexicons/api/types/site/standard/publication.ts @@ -15,7 +15,7 @@ const id = 'site.standard.publication' export interface Record { $type: 'site.standard.publication' basicTheme?: SiteStandardThemeBasic.Main - theme?: PubLeafletPublication.Theme + theme?: $Typed | { $type: string } description?: string icon?: BlobRef name: string diff --git a/lexicons/site/standard/publication.json b/lexicons/site/standard/publication.json index e1c3aea5..b1449052 100644 --- a/lexicons/site/standard/publication.json +++ b/lexicons/site/standard/publication.json @@ -9,8 +9,8 @@ "type": "ref" }, "theme": { - "type": "ref", - "ref": "pub.leaflet.publication#theme" + "type": "union", + "refs": ["pub.leaflet.publication#theme"] }, "description": { "maxGraphemes": 300, diff --git a/lexicons/src/normalize.ts b/lexicons/src/normalize.ts index 9b642c3f..d4890e07 100644 --- a/lexicons/src/normalize.ts +++ b/lexicons/src/normalize.ts @@ -14,7 +14,7 @@ */ import type * as PubLeafletDocument from "../api/types/pub/leaflet/document"; -import type * as PubLeafletPublication from "../api/types/pub/leaflet/publication"; +import * as PubLeafletPublication from "../api/types/pub/leaflet/publication"; import type * as PubLeafletContent from "../api/types/pub/leaflet/content"; import type * as SiteStandardDocument from "../api/types/site/standard/document"; import type * as SiteStandardPublication from "../api/types/site/standard/publication"; @@ -31,7 +31,20 @@ export type NormalizedDocument = SiteStandardDocument.Record & { }; // Normalized publication type - uses the generated site.standard.publication type -export type NormalizedPublication = SiteStandardPublication.Record; +// with the theme narrowed to only the valid pub.leaflet.publication#theme type +// (isTheme validates that $type is present, so we use $Typed) +// Note: We explicitly list fields rather than using Omit because the generated Record type +// has an index signature [k: string]: unknown that interferes with property typing +export type NormalizedPublication = { + $type: "site.standard.publication"; + name: string; + url: string; + description?: string; + icon?: SiteStandardPublication.Record["icon"]; + basicTheme?: SiteStandardThemeBasic.Main; + theme?: $Typed; + preferences?: SiteStandardPublication.Preferences; +}; /** * Checks if the record is a pub.leaflet.document @@ -210,9 +223,16 @@ export function normalizePublication( ): NormalizedPublication | null { if (!record || typeof record !== "object") return null; - // Pass through site.standard records directly + // Pass through site.standard records directly, but validate the theme if (isStandardPublication(record)) { - return record; + // Validate theme - only keep if it's a valid pub.leaflet.publication#theme + const theme = PubLeafletPublication.isTheme(record.theme) + ? (record.theme as $Typed) + : undefined; + return { + ...record, + theme, + }; } if (isLeafletPublication(record)) { @@ -225,6 +245,21 @@ export function normalizePublication( const basicTheme = leafletThemeToBasicTheme(record.theme); + // Validate theme - only keep if it's a valid pub.leaflet.publication#theme with $type set + // For legacy records without $type, add it during normalization + let theme: $Typed | undefined; + if (record.theme) { + if (PubLeafletPublication.isTheme(record.theme)) { + theme = record.theme as $Typed; + } else { + // Legacy theme without $type - add it + theme = { + ...record.theme, + $type: "pub.leaflet.publication#theme", + }; + } + } + // Convert preferences to site.standard format (strip/replace $type) const preferences: SiteStandardPublication.Preferences | undefined = record.preferences @@ -243,7 +278,7 @@ export function normalizePublication( description: record.description, icon: record.icon, basicTheme, - theme: record.theme, + theme, preferences, }; } -- 2.51.2 From c1168189d23fdb07f8436de655c1e2ec71a9ebe9 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 16:02:24 -0500 Subject: [PATCH 03/12] add standard.site validation --- .../site.standard.publication/route.ts | 34 +++++++++++++++++++ app/p/[didOrHandle]/[rkey]/page.tsx | 11 +++--- 2 files changed, 41 insertions(+), 4 deletions(-) create mode 100644 app/lish/[did]/[publication]/.well-known/site.standard.publication/route.ts diff --git a/app/lish/[did]/[publication]/.well-known/site.standard.publication/route.ts b/app/lish/[did]/[publication]/.well-known/site.standard.publication/route.ts new file mode 100644 index 00000000..70b980bd --- /dev/null +++ b/app/lish/[did]/[publication]/.well-known/site.standard.publication/route.ts @@ -0,0 +1,34 @@ +import { publicationNameOrUriFilter } from "src/utils/uriHelpers"; +import { supabaseServerClient } from "supabase/serverClient"; + +export async function GET( + req: Request, + props: { + params: Promise<{ publication: string; did: string }>; + }, +) { + let params = await props.params; + let did = decodeURIComponent(params.did); + let publication_name = decodeURIComponent(params.publication); + let [{ data: publications }] = await Promise.all([ + supabaseServerClient + .from("publications") + .select( + `*, + publication_subscriptions(*), + documents_in_publications(documents( + *, + comments_on_documents(count), + document_mentions_in_bsky(count) + )) + `, + ) + .eq("identity_did", did) + .or(publicationNameOrUriFilter(did, publication_name)) + .order("uri", { ascending: false }) + .limit(1), + ]); + let publication = publications?.[0]; + if (!did || !publication) return new Response(null, { status: 404 }); + return new Response(publication.uri); +} diff --git a/app/p/[didOrHandle]/[rkey]/page.tsx b/app/p/[didOrHandle]/[rkey]/page.tsx index ea7cf8fc..000f64ba 100644 --- a/app/p/[didOrHandle]/[rkey]/page.tsx +++ b/app/p/[didOrHandle]/[rkey]/page.tsx @@ -38,10 +38,13 @@ export async function generateMetadata(props: { return { icons: { - other: { - rel: "alternate", - url: document.uri, - }, + other: [ + { + rel: "alternate", + url: document.uri, + }, + { rel: "site.standard.document", url: document.uri }, + ], }, title: docRecord.title, description: docRecord?.description || "", -- 2.51.2 From bf14e1f768370534b5ca1be0a7ba3cc5d6649930 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 16:11:57 -0500 Subject: [PATCH 04/12] add link rel to pub post pages --- app/lish/[did]/[publication]/[rkey]/page.tsx | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/app/lish/[did]/[publication]/[rkey]/page.tsx b/app/lish/[did]/[publication]/[rkey]/page.tsx index 6bab19df..43739aa2 100644 --- a/app/lish/[did]/[publication]/[rkey]/page.tsx +++ b/app/lish/[did]/[publication]/[rkey]/page.tsx @@ -35,10 +35,13 @@ export async function generateMetadata(props: { sizes: "32x32", type: "image/png", }, - other: { - rel: "alternate", - url: document.uri, - }, + other: [ + { + rel: "alternate", + url: document.uri, + }, + { rel: "site.standard.document", url: document.uri }, + ], }, title: docRecord.title + -- 2.51.2 From 265dc693c39e91da2aa80472169e4b92db983ce0 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 16:30:54 -0500 Subject: [PATCH 05/12] add metadata column to identities table --- .../migrations/20260123000000_add_metadata_to_identities.sql | 1 + 1 file changed, 1 insertion(+) create mode 100644 supabase/migrations/20260123000000_add_metadata_to_identities.sql diff --git a/supabase/migrations/20260123000000_add_metadata_to_identities.sql b/supabase/migrations/20260123000000_add_metadata_to_identities.sql new file mode 100644 index 00000000..1e7b1fb3 --- /dev/null +++ b/supabase/migrations/20260123000000_add_metadata_to_identities.sql @@ -0,0 +1 @@ +alter table "public"."identities" add column "metadata" jsonb; -- 2.51.2 From 6bae920a82d4f0365b10b3b18e4f7efa6f43bb33 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 16:43:35 -0500 Subject: [PATCH 06/12] handle migrating users --- .../functions/migrate_user_to_standard.ts | 35 ++++++++++++++++--- app/api/oauth/[route]/route.ts | 11 ++++++ drizzle/schema.ts | 1 + supabase/database.types.ts | 3 ++ 4 files changed, 46 insertions(+), 4 deletions(-) diff --git a/app/api/inngest/functions/migrate_user_to_standard.ts b/app/api/inngest/functions/migrate_user_to_standard.ts index cf809b64..fc73f4c6 100644 --- a/app/api/inngest/functions/migrate_user_to_standard.ts +++ b/app/api/inngest/functions/migrate_user_to_standard.ts @@ -44,16 +44,33 @@ export const migrate_user_to_standard = inngest.createFunction( }; // Step 1: Verify OAuth session is valid - await step.run("verify-oauth-session", async () => { + const oauthValid = await step.run("verify-oauth-session", async () => { const result = await restoreOAuthSession(did); if (!result.ok) { - throw new Error( - `Failed to restore OAuth session: ${result.error.message}`, - ); + // Mark identity as needing migration so we can retry later + await supabaseServerClient + .from("identities") + .update({ + metadata: { needsStandardSiteMigration: true }, + }) + .eq("atp_did", did); + + return { success: false, error: result.error.message }; } return { success: true }; }); + if (!oauthValid.success) { + return { + success: false, + error: `Failed to restore OAuth session`, + stats, + publicationUriMap: {}, + documentUriMap: {}, + userSubscriptionUriMap: {}, + }; + } + // Step 2: Get user's pub.leaflet.publication records const oldPublications = await step.run( "fetch-old-publications", @@ -472,6 +489,16 @@ export const migrate_user_to_standard = inngest.createFunction( // 3. The normalization layer handles both schemas transparently for reads // Old records are also kept on the user's PDS so existing AT-URI references remain valid. + // Clear the migration flag on success + if (stats.errors.length === 0) { + await step.run("clear-migration-flag", async () => { + await supabaseServerClient + .from("identities") + .update({ metadata: null }) + .eq("atp_did", did); + }); + } + return { success: stats.errors.length === 0, stats, diff --git a/app/api/oauth/[route]/route.ts b/app/api/oauth/[route]/route.ts index 1cd296e4..903ce627 100644 --- a/app/api/oauth/[route]/route.ts +++ b/app/api/oauth/[route]/route.ts @@ -11,6 +11,7 @@ import { ActionAfterSignIn, parseActionFromSearchParam, } from "./afterSignInActions"; +import { inngest } from "app/api/inngest/client"; type OauthRequestClientState = { redirect: string | null; @@ -84,6 +85,16 @@ export async function GET( .single(); identity = data; } + + // Trigger migration if identity needs it + const metadata = identity?.metadata as Record | null; + if (metadata?.needsStandardSiteMigration) { + await inngest.send({ + name: "user/migrate-to-standard", + data: { did: session.did }, + }); + } + let { data: token } = await supabaseServerClient .from("email_auth_tokens") .insert({ diff --git a/drizzle/schema.ts b/drizzle/schema.ts index 6e1f7f71..65cd9add 100644 --- a/drizzle/schema.ts +++ b/drizzle/schema.ts @@ -140,6 +140,7 @@ export const identities = pgTable("identities", { email: text("email"), atp_did: text("atp_did"), interface_state: jsonb("interface_state"), + metadata: jsonb("metadata"), }, (table) => { return { diff --git a/supabase/database.types.ts b/supabase/database.types.ts index 757213b4..a385b699 100644 --- a/supabase/database.types.ts +++ b/supabase/database.types.ts @@ -551,6 +551,7 @@ export type Database = { home_page: string id: string interface_state: Json | null + metadata: Json | null } Insert: { atp_did?: string | null @@ -559,6 +560,7 @@ export type Database = { home_page?: string id?: string interface_state?: Json | null + metadata?: Json | null } Update: { atp_did?: string | null @@ -567,6 +569,7 @@ export type Database = { home_page?: string id?: string interface_state?: Json | null + metadata?: Json | null } Relationships: [ { -- 2.51.2 From 4b004e0c04c3bddc49fe1b695a105fdcd3bdf400 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 17:29:30 -0500 Subject: [PATCH 07/12] add inngest workflow to cleanup expired oauth sessions --- app/api/inngest/client.ts | 3 + .../cleanup_expired_oauth_sessions.ts | 101 ++++++++++++++++++ app/api/inngest/route.tsx | 2 + 3 files changed, 106 insertions(+) create mode 100644 app/api/inngest/functions/cleanup_expired_oauth_sessions.ts diff --git a/app/api/inngest/client.ts b/app/api/inngest/client.ts index 77a2266a..caf5854f 100644 --- a/app/api/inngest/client.ts +++ b/app/api/inngest/client.ts @@ -26,6 +26,9 @@ export type Events = { did: string; }; }; + "user/cleanup-expired-oauth-sessions": { + data: {}; + }; }; // Create a client to send and receive events diff --git a/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts b/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts new file mode 100644 index 00000000..65cf814e --- /dev/null +++ b/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts @@ -0,0 +1,101 @@ +import { supabaseServerClient } from "supabase/serverClient"; +import { inngest } from "../client"; +import { restoreOAuthSession } from "src/atproto-oauth"; + +export const cleanup_expired_oauth_sessions = inngest.createFunction( + { id: "cleanup_expired_oauth_sessions" }, + { event: "user/cleanup-expired-oauth-sessions" }, + async ({ step }) => { + const stats = { + totalIdentities: 0, + validSessions: 0, + expiredSessions: 0, + tokensDeleted: 0, + errors: [] as string[], + }; + + // Step 1: Get all identities with an atp_did (OAuth users) + const identities = await step.run("fetch-oauth-identities", async () => { + const { data, error } = await supabaseServerClient + .from("identities") + .select("id, atp_did") + .not("atp_did", "is", null); + + if (error) { + throw new Error(`Failed to fetch identities: ${error.message}`); + } + return data || []; + }); + + stats.totalIdentities = identities.length; + console.log(`Found ${identities.length} OAuth identities to check`); + + // Step 2: Check each identity's OAuth session and cleanup if expired + for (const identity of identities) { + if (!identity.atp_did) continue; + + const result = await step.run( + `check-session-${identity.id}`, + async () => { + console.log(`Checking OAuth session for DID: ${identity.atp_did}`); + + const sessionResult = await restoreOAuthSession(identity.atp_did!); + + if (sessionResult.ok) { + console.log(` Session valid for ${identity.atp_did}`); + return { valid: true, tokensDeleted: 0 }; + } + + // Session is expired/invalid - delete associated auth tokens + console.log( + ` Session expired for ${identity.atp_did}: ${sessionResult.error.message}`, + ); + + const { data: deletedTokens, error: deleteError } = + await supabaseServerClient + .from("email_auth_tokens") + .delete() + .eq("identity", identity.id) + .select("id"); + + if (deleteError) { + console.error( + ` Error deleting tokens for identity ${identity.id}: ${deleteError.message}`, + ); + return { + valid: false, + tokensDeleted: 0, + error: deleteError.message, + }; + } + + const deletedCount = deletedTokens?.length || 0; + console.log( + ` Deleted ${deletedCount} auth tokens for identity ${identity.id}`, + ); + + return { valid: false, tokensDeleted: deletedCount }; + }, + ); + + if (result.valid) { + stats.validSessions++; + } else { + stats.expiredSessions++; + stats.tokensDeleted += result.tokensDeleted; + if ("error" in result && result.error) { + stats.errors.push( + `Identity ${identity.id}: ${result.error}`, + ); + } + } + } + + console.log("Cleanup completed:", stats); + + return { + success: stats.errors.length === 0, + stats, + }; + }, +); diff --git a/app/api/inngest/route.tsx b/app/api/inngest/route.tsx index f8886099..37fc24c4 100644 --- a/app/api/inngest/route.tsx +++ b/app/api/inngest/route.tsx @@ -5,6 +5,7 @@ import { come_online } from "./functions/come_online"; import { batched_update_profiles } from "./functions/batched_update_profiles"; import { index_follows } from "./functions/index_follows"; import { migrate_user_to_standard } from "./functions/migrate_user_to_standard"; +import { cleanup_expired_oauth_sessions } from "./functions/cleanup_expired_oauth_sessions"; export const { GET, POST, PUT } = serve({ client: inngest, @@ -14,5 +15,6 @@ export const { GET, POST, PUT } = serve({ batched_update_profiles, index_follows, migrate_user_to_standard, + cleanup_expired_oauth_sessions, ], }); -- 2.51.2 From c740c60fe0bff9d70abff649e74404abf48dcba7 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 17:40:33 -0500 Subject: [PATCH 08/12] update oauth cleanup workflow --- .../cleanup_expired_oauth_sessions.ts | 119 +++++++++++------- 1 file changed, 76 insertions(+), 43 deletions(-) diff --git a/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts b/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts index 65cf814e..3da705a4 100644 --- a/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts +++ b/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts @@ -14,79 +14,112 @@ export const cleanup_expired_oauth_sessions = inngest.createFunction( errors: [] as string[], }; - // Step 1: Get all identities with an atp_did (OAuth users) + // Step 1: Get all identities with an atp_did (OAuth users) that have at least one auth token const identities = await step.run("fetch-oauth-identities", async () => { const { data, error } = await supabaseServerClient .from("identities") - .select("id, atp_did") + .select("id, atp_did, email_auth_tokens(count)") .not("atp_did", "is", null); if (error) { throw new Error(`Failed to fetch identities: ${error.message}`); } - return data || []; + + // Filter to only include identities with at least one auth token + return (data || []) + .filter((identity) => { + const tokenCount = identity.email_auth_tokens?.[0]?.count ?? 0; + return tokenCount > 0; + }) + .map((identity) => ({ + id: identity.id, + atp_did: identity.atp_did, + tokenCount: identity.email_auth_tokens?.[0]?.count ?? 0, + })); }); stats.totalIdentities = identities.length; - console.log(`Found ${identities.length} OAuth identities to check`); - - // Step 2: Check each identity's OAuth session and cleanup if expired - for (const identity of identities) { - if (!identity.atp_did) continue; + console.log(`Found ${identities.length} OAuth identities with active sessions to check`); + + // Step 2: Check identities' OAuth sessions in batched parallel and cleanup if expired + const BATCH_SIZE = 150; + const allResults: { + identityId: string; + valid: boolean; + tokensDeleted: number; + error?: string; + }[] = []; + + for (let i = 0; i < identities.length; i += BATCH_SIZE) { + const batch = identities.slice(i, i + BATCH_SIZE); + const batchNum = Math.floor(i / BATCH_SIZE) + 1; + const totalBatches = Math.ceil(identities.length / BATCH_SIZE); + + console.log( + `Processing batch ${batchNum}/${totalBatches} (${batch.length} identities)`, + ); - const result = await step.run( - `check-session-${identity.id}`, - async () => { - console.log(`Checking OAuth session for DID: ${identity.atp_did}`); + const batchResults = await Promise.all( + batch.map((identity) => + step.run(`check-session-${identity.id}`, async () => { + console.log( + `Checking OAuth session for DID: ${identity.atp_did} (${identity.tokenCount} tokens)`, + ); - const sessionResult = await restoreOAuthSession(identity.atp_did!); + const sessionResult = await restoreOAuthSession(identity.atp_did!); - if (sessionResult.ok) { - console.log(` Session valid for ${identity.atp_did}`); - return { valid: true, tokensDeleted: 0 }; - } + if (sessionResult.ok) { + console.log(` Session valid for ${identity.atp_did}`); + return { identityId: identity.id, valid: true, tokensDeleted: 0 }; + } - // Session is expired/invalid - delete associated auth tokens - console.log( - ` Session expired for ${identity.atp_did}: ${sessionResult.error.message}`, - ); + // Session is expired/invalid - delete associated auth tokens + console.log( + ` Session expired for ${identity.atp_did}: ${sessionResult.error.message}`, + ); - const { data: deletedTokens, error: deleteError } = - await supabaseServerClient + const { error: deleteError } = await supabaseServerClient .from("email_auth_tokens") .delete() - .eq("identity", identity.id) - .select("id"); - - if (deleteError) { - console.error( - ` Error deleting tokens for identity ${identity.id}: ${deleteError.message}`, + .eq("identity", identity.id); + + if (deleteError) { + console.error( + ` Error deleting tokens for identity ${identity.id}: ${deleteError.message}`, + ); + return { + identityId: identity.id, + valid: false, + tokensDeleted: 0, + error: deleteError.message, + }; + } + + console.log( + ` Deleted ${identity.tokenCount} auth tokens for identity ${identity.id}`, ); + return { + identityId: identity.id, valid: false, - tokensDeleted: 0, - error: deleteError.message, + tokensDeleted: identity.tokenCount, }; - } - - const deletedCount = deletedTokens?.length || 0; - console.log( - ` Deleted ${deletedCount} auth tokens for identity ${identity.id}`, - ); - - return { valid: false, tokensDeleted: deletedCount }; - }, + }), + ), ); + allResults.push(...batchResults); + } + + // Aggregate results + for (const result of allResults) { if (result.valid) { stats.validSessions++; } else { stats.expiredSessions++; stats.tokensDeleted += result.tokensDeleted; if ("error" in result && result.error) { - stats.errors.push( - `Identity ${identity.id}: ${result.error}`, - ); + stats.errors.push(`Identity ${result.identityId}: ${result.error}`); } } } -- 2.51.2 From fb84b49333633f63872c4cf0b3ba60244d587810 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 17:58:19 -0500 Subject: [PATCH 09/12] update check oauth session workflow --- app/api/inngest/client.ts | 7 + .../cleanup_expired_oauth_sessions.ts | 165 ++++++++---------- app/api/inngest/route.tsx | 6 +- 3 files changed, 89 insertions(+), 89 deletions(-) diff --git a/app/api/inngest/client.ts b/app/api/inngest/client.ts index caf5854f..4f082923 100644 --- a/app/api/inngest/client.ts +++ b/app/api/inngest/client.ts @@ -29,6 +29,13 @@ export type Events = { "user/cleanup-expired-oauth-sessions": { data: {}; }; + "user/check-oauth-session": { + data: { + identityId: string; + did: string; + tokenCount: number; + }; + }; }; // Create a client to send and receive events diff --git a/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts b/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts index 3da705a4..486cee15 100644 --- a/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts +++ b/app/api/inngest/functions/cleanup_expired_oauth_sessions.ts @@ -2,19 +2,12 @@ import { supabaseServerClient } from "supabase/serverClient"; import { inngest } from "../client"; import { restoreOAuthSession } from "src/atproto-oauth"; +// Main function that fetches identities and publishes events for each one export const cleanup_expired_oauth_sessions = inngest.createFunction( { id: "cleanup_expired_oauth_sessions" }, { event: "user/cleanup-expired-oauth-sessions" }, async ({ step }) => { - const stats = { - totalIdentities: 0, - validSessions: 0, - expiredSessions: 0, - tokensDeleted: 0, - errors: [] as string[], - }; - - // Step 1: Get all identities with an atp_did (OAuth users) that have at least one auth token + // Get all identities with an atp_did (OAuth users) that have at least one auth token const identities = await step.run("fetch-oauth-identities", async () => { const { data, error } = await supabaseServerClient .from("identities") @@ -33,102 +26,98 @@ export const cleanup_expired_oauth_sessions = inngest.createFunction( }) .map((identity) => ({ id: identity.id, - atp_did: identity.atp_did, + atp_did: identity.atp_did!, tokenCount: identity.email_auth_tokens?.[0]?.count ?? 0, })); }); - stats.totalIdentities = identities.length; - console.log(`Found ${identities.length} OAuth identities with active sessions to check`); + console.log( + `Found ${identities.length} OAuth identities with active sessions to check`, + ); - // Step 2: Check identities' OAuth sessions in batched parallel and cleanup if expired - const BATCH_SIZE = 150; - const allResults: { - identityId: string; - valid: boolean; - tokensDeleted: number; - error?: string; - }[] = []; + // Publish events for each identity in batches + const BATCH_SIZE = 100; + let totalSent = 0; for (let i = 0; i < identities.length; i += BATCH_SIZE) { const batch = identities.slice(i, i + BATCH_SIZE); - const batchNum = Math.floor(i / BATCH_SIZE) + 1; - const totalBatches = Math.ceil(identities.length / BATCH_SIZE); - console.log( - `Processing batch ${batchNum}/${totalBatches} (${batch.length} identities)`, - ); + await step.run(`send-events-batch-${i}`, async () => { + const events = batch.map((identity) => ({ + name: "user/check-oauth-session" as const, + data: { + identityId: identity.id, + did: identity.atp_did, + tokenCount: identity.tokenCount, + }, + })); - const batchResults = await Promise.all( - batch.map((identity) => - step.run(`check-session-${identity.id}`, async () => { - console.log( - `Checking OAuth session for DID: ${identity.atp_did} (${identity.tokenCount} tokens)`, - ); - - const sessionResult = await restoreOAuthSession(identity.atp_did!); - - if (sessionResult.ok) { - console.log(` Session valid for ${identity.atp_did}`); - return { identityId: identity.id, valid: true, tokensDeleted: 0 }; - } - - // Session is expired/invalid - delete associated auth tokens - console.log( - ` Session expired for ${identity.atp_did}: ${sessionResult.error.message}`, - ); - - const { error: deleteError } = await supabaseServerClient - .from("email_auth_tokens") - .delete() - .eq("identity", identity.id); - - if (deleteError) { - console.error( - ` Error deleting tokens for identity ${identity.id}: ${deleteError.message}`, - ); - return { - identityId: identity.id, - valid: false, - tokensDeleted: 0, - error: deleteError.message, - }; - } - - console.log( - ` Deleted ${identity.tokenCount} auth tokens for identity ${identity.id}`, - ); - - return { - identityId: identity.id, - valid: false, - tokensDeleted: identity.tokenCount, - }; - }), - ), - ); + await inngest.send(events); + return events.length; + }); - allResults.push(...batchResults); + totalSent += batch.length; } - // Aggregate results - for (const result of allResults) { - if (result.valid) { - stats.validSessions++; - } else { - stats.expiredSessions++; - stats.tokensDeleted += result.tokensDeleted; - if ("error" in result && result.error) { - stats.errors.push(`Identity ${result.identityId}: ${result.error}`); - } + console.log(`Published ${totalSent} check-oauth-session events`); + + return { + success: true, + identitiesQueued: totalSent, + }; + }, +); + +// Function that checks a single identity's OAuth session and cleans up if expired +export const check_oauth_session = inngest.createFunction( + { id: "check_oauth_session" }, + { event: "user/check-oauth-session" }, + async ({ event, step }) => { + const { identityId, did, tokenCount } = event.data; + + const result = await step.run("check-and-cleanup", async () => { + console.log(`Checking OAuth session for DID: ${did} (${tokenCount} tokens)`); + + const sessionResult = await restoreOAuthSession(did); + + if (sessionResult.ok) { + console.log(` Session valid for ${did}`); + return { valid: true, tokensDeleted: 0 }; } - } - console.log("Cleanup completed:", stats); + // Session is expired/invalid - delete associated auth tokens + console.log( + ` Session expired for ${did}: ${sessionResult.error.message}`, + ); + + const { error: deleteError } = await supabaseServerClient + .from("email_auth_tokens") + .delete() + .eq("identity", identityId); + + if (deleteError) { + console.error( + ` Error deleting tokens for identity ${identityId}: ${deleteError.message}`, + ); + return { + valid: false, + tokensDeleted: 0, + error: deleteError.message, + }; + } + + console.log(` Deleted ${tokenCount} auth tokens for identity ${identityId}`); + + return { + valid: false, + tokensDeleted: tokenCount, + }; + }); return { - success: stats.errors.length === 0, - stats, + identityId, + did, + ...result, }; }, ); diff --git a/app/api/inngest/route.tsx b/app/api/inngest/route.tsx index 37fc24c4..e086dc2c 100644 --- a/app/api/inngest/route.tsx +++ b/app/api/inngest/route.tsx @@ -5,7 +5,10 @@ import { come_online } from "./functions/come_online"; import { batched_update_profiles } from "./functions/batched_update_profiles"; import { index_follows } from "./functions/index_follows"; import { migrate_user_to_standard } from "./functions/migrate_user_to_standard"; -import { cleanup_expired_oauth_sessions } from "./functions/cleanup_expired_oauth_sessions"; +import { + cleanup_expired_oauth_sessions, + check_oauth_session, +} from "./functions/cleanup_expired_oauth_sessions"; export const { GET, POST, PUT } = serve({ client: inngest, @@ -16,5 +19,6 @@ export const { GET, POST, PUT } = serve({ index_follows, migrate_user_to_standard, cleanup_expired_oauth_sessions, + check_oauth_session, ], }); -- 2.51.2 From c2ed386ab88cb5c9948eee62866c2383f094731d Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 20:32:47 -0500 Subject: [PATCH 10/12] update migration workflow to handle old/new records --- .../functions/migrate_user_to_standard.ts | 125 ++++++++++++++++++ 1 file changed, 125 insertions(+) diff --git a/app/api/inngest/functions/migrate_user_to_standard.ts b/app/api/inngest/functions/migrate_user_to_standard.ts index fc73f4c6..afc51931 100644 --- a/app/api/inngest/functions/migrate_user_to_standard.ts +++ b/app/api/inngest/functions/migrate_user_to_standard.ts @@ -38,6 +38,7 @@ export const migrate_user_to_standard = inngest.createFunction( const stats = { publicationsMigrated: 0, documentsMigrated: 0, + standardDocumentsFixed: 0, userSubscriptionsMigrated: 0, referencesUpdated: 0, errors: [] as string[], @@ -323,6 +324,130 @@ export const migrate_user_to_standard = inngest.createFunction( } } + // Step 4b: Fix existing site.standard.document records that reference pub.leaflet.publication + // This handles the case where site.standard.document records were created pointing to + // pub.leaflet.publication URIs before the publication was migrated to site.standard.publication + const existingStandardDocs = await step.run( + "fetch-existing-standard-documents", + async () => { + const { data, error } = await supabaseServerClient + .from("documents") + .select("uri, data") + .like("uri", `at://${did}/site.standard.document/%`); + + if (error) + throw new Error( + `Failed to fetch existing standard documents: ${error.message}`, + ); + return data || []; + }, + ); + + // Find documents that reference pub.leaflet.publication and need their site field updated + const standardDocsToFix = existingStandardDocs + .map((doc) => { + const data = doc.data as SiteStandardDocument.Record; + const site = data?.site; + + // Check if site field references a pub.leaflet.publication + if (!site || !site.includes("/pub.leaflet.publication/")) { + return null; + } + + try { + const oldPubAturi = new AtUri(site); + const newPubUri = `at://${oldPubAturi.hostname}/site.standard.publication/${oldPubAturi.rkey}`; + + // Only fix if we have the new publication in our map (meaning it was migrated) + // or if the new publication exists (check against all migrated publications) + if ( + publicationUriMap[site] || + Object.values(publicationUriMap).includes(newPubUri) + ) { + const docAturi = new AtUri(doc.uri); + const updatedRecord: SiteStandardDocument.Record = { + ...data, + site: newPubUri, + }; + + return { + doc, + rkey: docAturi.rkey, + oldSite: site, + newSite: newPubUri, + updatedRecord, + }; + } + } catch (e) { + stats.errors.push(`Invalid site URI in document ${doc.uri}: ${site}`); + } + + return null; + }) + .filter((x) => x !== null); + + // Update these documents on PDS and in database + if (standardDocsToFix.length > 0) { + const fixResults = await Promise.all( + standardDocsToFix.map(({ doc, rkey, oldSite, newSite, updatedRecord }) => + step.run(`fix-standard-document-${doc.uri}`, async () => { + // PDS write to update the site field + const agent = await createAuthenticatedAgent(did); + await agent.com.atproto.repo.putRecord({ + repo: did, + collection: "site.standard.document", + rkey, + record: updatedRecord, + validate: false, + }); + + // DB write + const { error: dbError } = await supabaseServerClient + .from("documents") + .update({ data: updatedRecord as Json }) + .eq("uri", doc.uri); + + if (dbError) { + return { + success: false as const, + uri: doc.uri, + error: dbError.message, + }; + } + + // Update documents_in_publications to point to new publication URI + await supabaseServerClient + .from("documents_in_publications") + .upsert({ + publication: newSite, + document: doc.uri, + }); + + // Remove old publication reference if different + if (oldSite !== newSite) { + await supabaseServerClient + .from("documents_in_publications") + .delete() + .eq("publication", oldSite) + .eq("document", doc.uri); + } + + return { success: true as const, uri: doc.uri }; + }), + ), + ); + + for (const result of fixResults) { + if (result.success) { + stats.standardDocumentsFixed++; + } else { + stats.errors.push( + `Fix standard document ${result.uri}: Database error: ${result.error}`, + ); + } + } + } + // Step 5: Update references in database tables (all in parallel) await step.run("update-references", async () => { const pubEntries = Object.entries(publicationUriMap); -- 2.51.2 From 25afb91c8b80747b2a1d9fea1f51ac44899bf523 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 21:15:41 -0500 Subject: [PATCH 11/12] set theme properly on update --- app/lish/createPub/updatePublication.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/app/lish/createPub/updatePublication.ts b/app/lish/createPub/updatePublication.ts index 81a87aa2..a9bd3e56 100644 --- a/app/lish/createPub/updatePublication.ts +++ b/app/lish/createPub/updatePublication.ts @@ -278,6 +278,7 @@ export async function updatePublicationTheme({ return withPublicationUpdate(uri, async ({ normalizedPub, existingBasePath, publicationType, agent }) => { // Build theme object const themeData = { + $type: "pub.leaflet.publication#theme" as const, backgroundImage: theme.backgroundImage ? { $type: "pub.leaflet.theme.backgroundImage", -- 2.51.2 From ff2ed11377b2273b278a697324530a31bd96cdd2 Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Fri, 23 Jan 2026 22:51:06 -0500 Subject: [PATCH 12/12] add new fix inngest function --- app/api/inngest/client.ts | 5 + .../fix_standard_document_publications.ts | 213 ++++++++++++++++++ app/api/inngest/route.tsx | 2 + 3 files changed, 220 insertions(+) create mode 100644 app/api/inngest/functions/fix_standard_document_publications.ts diff --git a/app/api/inngest/client.ts b/app/api/inngest/client.ts index 4f082923..27dfa5ac 100644 --- a/app/api/inngest/client.ts +++ b/app/api/inngest/client.ts @@ -36,6 +36,11 @@ export type Events = { tokenCount: number; }; }; + "documents/fix-publication-references": { + data: { + documentUris: string[]; + }; + }; }; // Create a client to send and receive events diff --git a/app/api/inngest/functions/fix_standard_document_publications.ts b/app/api/inngest/functions/fix_standard_document_publications.ts new file mode 100644 index 00000000..8b1657e3 --- /dev/null +++ b/app/api/inngest/functions/fix_standard_document_publications.ts @@ -0,0 +1,213 @@ +import { supabaseServerClient } from "supabase/serverClient"; +import { inngest } from "../client"; +import { restoreOAuthSession } from "src/atproto-oauth"; +import { AtpBaseClient, SiteStandardDocument } from "lexicons/api"; +import { AtUri } from "@atproto/syntax"; +import { Json } from "supabase/database.types"; + +async function createAuthenticatedAgent(did: string): Promise { + const result = await restoreOAuthSession(did); + if (!result.ok) { + throw new Error(`Failed to restore OAuth session: ${result.error.message}`); + } + const credentialSession = result.value; + return new AtpBaseClient( + credentialSession.fetchHandler.bind(credentialSession), + ); +} + +/** + * Fixes site.standard.document records that have stale pub.leaflet.publication + * references in their site field. Updates both the PDS record and database. + */ +export const fix_standard_document_publications = inngest.createFunction( + { id: "fix_standard_document_publications" }, + { event: "documents/fix-publication-references" }, + async ({ event, step }) => { + const { documentUris } = event.data as { documentUris: string[] }; + + const stats = { + documentsFixed: 0, + joinEntriesFixed: 0, + errors: [] as string[], + }; + + if (!documentUris || documentUris.length === 0) { + return { success: true, stats, message: "No documents to fix" }; + } + + // Group documents by DID (author) for efficient OAuth session handling + const docsByDid = new Map(); + for (const uri of documentUris) { + try { + const aturi = new AtUri(uri); + const did = aturi.hostname; + const existing = docsByDid.get(did) || []; + existing.push(uri); + docsByDid.set(did, existing); + } catch (e) { + stats.errors.push(`Invalid URI: ${uri}`); + } + } + + // Process each DID's documents + for (const [did, uris] of docsByDid) { + // Verify OAuth session for this user + const oauthValid = await step.run( + `verify-oauth-${did.slice(-8)}`, + async () => { + const result = await restoreOAuthSession(did); + return result.ok; + }, + ); + + if (!oauthValid) { + stats.errors.push(`No valid OAuth session for ${did}`); + continue; + } + + // Fix each document + for (const docUri of uris) { + const result = await step.run( + `fix-doc-${docUri.slice(-12)}`, + async () => { + // Fetch the document + const { data: doc, error: fetchError } = await supabaseServerClient + .from("documents") + .select("uri, data") + .eq("uri", docUri) + .single(); + + if (fetchError || !doc) { + return { + success: false as const, + error: `Document not found: ${fetchError?.message || "no data"}`, + }; + } + + const data = doc.data as SiteStandardDocument.Record; + const oldSite = data?.site; + + if (!oldSite || !oldSite.includes("/pub.leaflet.publication/")) { + return { + success: false as const, + error: "Document does not have a pub.leaflet.publication site reference", + }; + } + + // Convert to new publication URI + const oldPubAturi = new AtUri(oldSite); + const newSite = `at://${oldPubAturi.hostname}/site.standard.publication/${oldPubAturi.rkey}`; + + // Update the record + const updatedRecord: SiteStandardDocument.Record = { + ...data, + site: newSite, + }; + + // Write to PDS + const docAturi = new AtUri(docUri); + const agent = await createAuthenticatedAgent(did); + await agent.com.atproto.repo.putRecord({ + repo: did, + collection: "site.standard.document", + rkey: docAturi.rkey, + record: updatedRecord, + validate: false, + }); + + // Update database + const { error: dbError } = await supabaseServerClient + .from("documents") + .update({ data: updatedRecord as Json }) + .eq("uri", docUri); + + if (dbError) { + return { + success: false as const, + error: `Database update failed: ${dbError.message}`, + }; + } + + return { + success: true as const, + oldSite, + newSite, + }; + }, + ); + + if (result.success) { + stats.documentsFixed++; + + // Fix the documents_in_publications entry + const joinResult = await step.run( + `fix-join-${docUri.slice(-12)}`, + async () => { + // Find the publication URI that exists in the database + const { data: doc } = await supabaseServerClient + .from("documents") + .select("data") + .eq("uri", docUri) + .single(); + + const newSite = (doc?.data as any)?.site; + if (!newSite) { + return { success: false as const, error: "Could not read updated document" }; + } + + // Check which publication URI exists + const newPubAturi = new AtUri(newSite); + const oldPubUri = `at://${newPubAturi.hostname}/pub.leaflet.publication/${newPubAturi.rkey}`; + + const { data: pubs } = await supabaseServerClient + .from("publications") + .select("uri") + .in("uri", [newSite, oldPubUri]); + + const existingPubUri = pubs?.find((p) => p.uri === newSite)?.uri || + pubs?.find((p) => p.uri === oldPubUri)?.uri; + + if (!existingPubUri) { + return { success: false as const, error: "No matching publication found" }; + } + + // Delete any existing entries for this document + await supabaseServerClient + .from("documents_in_publications") + .delete() + .eq("document", docUri); + + // Insert the correct entry + const { error: insertError } = await supabaseServerClient + .from("documents_in_publications") + .insert({ + document: docUri, + publication: existingPubUri, + }); + + if (insertError) { + return { success: false as const, error: insertError.message }; + } + + return { success: true as const, publication: existingPubUri }; + }, + ); + + if (joinResult.success) { + stats.joinEntriesFixed++; + } else { + stats.errors.push(`Join table fix failed for ${docUri}: ${"error" in joinResult ? joinResult.error : "unknown error"}`); + } + } else { + stats.errors.push(`${docUri}: ${result.error}`); + } + } + } + + return { + success: stats.errors.length === 0, + stats, + }; + }, +); diff --git a/app/api/inngest/route.tsx b/app/api/inngest/route.tsx index e086dc2c..b867e731 100644 --- a/app/api/inngest/route.tsx +++ b/app/api/inngest/route.tsx @@ -5,6 +5,7 @@ import { come_online } from "./functions/come_online"; import { batched_update_profiles } from "./functions/batched_update_profiles"; import { index_follows } from "./functions/index_follows"; import { migrate_user_to_standard } from "./functions/migrate_user_to_standard"; +import { fix_standard_document_publications } from "./functions/fix_standard_document_publications"; import { cleanup_expired_oauth_sessions, check_oauth_session, @@ -18,6 +19,7 @@ export const { GET, POST, PUT } = serve({ batched_update_profiles, index_follows, migrate_user_to_standard, + fix_standard_document_publications, cleanup_expired_oauth_sessions, check_oauth_session, ],