Feature/scheduled posts

actions/cancelScheduledPost.ts

@@ -0,0 +1,50 @@
+"use server";
+
+import { getIdentityData } from "actions/getIdentityData";
+import { supabaseServerClient } from "supabase/serverClient";
+import { Result, Ok, Err } from "src/result";
+import { updateScheduleColumns } from "actions/scheduledPublishDb";
+
+export type CancelScheduledPostError =
+  | { type: "not_authenticated" }
+  | { type: "not_found" };
+
+export async function cancelScheduledPost(args: {
+  leaflet_id: string;
+  publication_uri?: string;
+}): Promise<Result<{ cancelled: true }, CancelScheduledPostError>> {
+  let identity = await getIdentityData();
+  if (!identity || !identity.atp_did) return Err({ type: "not_authenticated" });
+
+  if (!identity.entitlements?.can_schedule_posts) {
+    return Err({ type: "not_found" });
+  }
+
+  if (args.publication_uri) {
+    const { data: pub } = await supabaseServerClient
+      .from("publications")
+      .select("identity_did")
+      .eq("uri", args.publication_uri)
+      .single();
+    if (!pub || pub.identity_did !== identity.atp_did) {
+      return Err({ type: "not_found" });
+    }
+  } else {
+    const { data: ownership } = await supabaseServerClient
+      .from("permission_token_on_homepage")
+      .select("token")
+      .eq("token", args.leaflet_id)
+      .eq("identity", identity.id)
+      .maybeSingle();
+    if (!ownership) return Err({ type: "not_found" });
+  }
+
+  const { found } = await updateScheduleColumns(
+    args.leaflet_id,
+    args.publication_uri,
+    { scheduled_publish_at: null, scheduled_publish_data: null },
+  );
+  if (!found) return Err({ type: "not_found" });
+
+  return Ok({ cancelled: true });
+}

actions/publishToPublication.ts

@@ -1,6 +1,7 @@
 "use server";
 
 import { restoreOAuthSession, OAuthSessionError } from "src/atproto-oauth";
+import { OAuthSession } from "@atproto/oauth-client-node";
 import { getIdentityData } from "actions/getIdentityData";
 import {
   AtpBaseClient,
@@ -49,18 +50,7 @@ type PublishResult =
   | { success: true; rkey: string; record: SiteStandardDocument.Record }
   | { success: false; error: OAuthSessionError };
 
-export async function publishToPublication({
-  root_entity,
-  publication_uri,
-  leaflet_id,
-  title,
-  description,
-  tags,
-  cover_image,
-  entitiesToDelete,
-  publishedAt,
-  postPreferences,
-}: {
+type PublishArgs = {
   root_entity: string;
   publication_uri?: string;
   leaflet_id: string;
@@ -75,7 +65,11 @@ export async function publishToPublication({
     showMentions?: boolean;
     showRecommends?: boolean;
   } | null;
-}): Promise<PublishResult> {
+};
+
+export async function publishToPublication(
+  args: PublishArgs,
+): Promise<PublishResult> {
   let identity = await getIdentityData();
   if (!identity || !identity.atp_did) {
     return {
@@ -92,7 +86,31 @@ export async function publishToPublication({
   if (!sessionResult.ok) {
     return { success: false, error: sessionResult.error };
   }
-  let credentialSession = sessionResult.value;
+
+  return publishToPublicationWithSession({
+    ...args,
+    credentialSession: sessionResult.value,
+    did: identity.atp_did,
+  });
+}
+
+export async function publishToPublicationWithSession({
+  root_entity,
+  publication_uri,
+  leaflet_id,
+  title,
+  description,
+  tags,
+  cover_image,
+  entitiesToDelete,
+  publishedAt,
+  postPreferences,
+  credentialSession,
+  did,
+}: PublishArgs & {
+  credentialSession: OAuthSession;
+  did: string;
+}): Promise<PublishResult> {
   let agent = new AtpBaseClient(
     credentialSession.fetchHandler.bind(credentialSession),
   );
@@ -111,7 +129,7 @@ export async function publishToPublication({
       .single();
     console.log(error);
 
-    if (!data || identity.atp_did !== data?.identity_did)
+    if (!data || did !== data?.identity_did)
       throw new Error("No draft or not publisher");
     draft = data.leaflets_in_publications[0];
     existingDocUri = draft?.doc;
@@ -128,13 +146,13 @@ export async function publishToPublication({
     // If updating an existing document, verify the current user is the owner
     if (existingDocUri) {
       let docOwner = new AtUri(existingDocUri).host;
-      if (docOwner !== identity.atp_did) {
+      if (docOwner !== did) {
         return {
           success: false,
           error: {
             type: "oauth_session_expired" as const,
             message: "Not the document owner",
-            did: identity.atp_did,
+            did,
           },
         };
       }
@@ -346,7 +364,9 @@ export async function publishToPublication({
   });
 
   if (publication_uri) {
-    // Publishing to a publication - update both tables
+    // Publishing to a publication - update both tables.
+    // Always clear scheduled_publish_* so a manual publish (or backdate) of a
+    // previously-scheduled post cancels the pending inngest run on wake-up.
     await Promise.all([
       supabaseServerClient.from("documents_in_publications").upsert({
         publication: publication_uri,
@@ -360,6 +380,8 @@ export async function publishToPublication({
         description: description,
         tags: resolvedTags ?? [],
         cover_image: cover_image ?? null,
+        scheduled_publish_at: null,
+        scheduled_publish_data: null,
       }),
     ]);
   } else {
@@ -371,6 +393,8 @@ export async function publishToPublication({
       description: description || "",
       tags: resolvedTags ?? [],
       cover_image: cover_image ?? null,
+      scheduled_publish_at: null,
+      scheduled_publish_data: null,
     });
 
     // Heuristic: Remove title entities if this is the first time publishing standalone

actions/schedulePost.ts

@@ -0,0 +1,116 @@
+"use server";
+
+import { AppBskyRichtextFacet } from "@atproto/api";
+import { getIdentityData } from "actions/getIdentityData";
+import { inngest } from "app/api/inngest/client";
+import { supabaseServerClient } from "supabase/serverClient";
+import { Json } from "supabase/database.types";
+import { Result, Ok, Err } from "src/result";
+import {
+  ScheduleUpdates,
+  updateScheduleColumns,
+} from "actions/scheduledPublishDb";
+import type { ScheduledPublishData } from "src/utils/scheduledPublish";
+
+export type SchedulePostError =
+  | { type: "not_authenticated" }
+  | { type: "not_pro" }
+  | { type: "invalid_schedule" }
+  | { type: "not_found" };
+
+export async function schedulePost(args: {
+  leaflet_id: string;
+  publication_uri?: string;
+  scheduled_publish_at: string;
+  title?: string;
+  description?: string;
+  tags?: string[];
+  cover_image?: string | null;
+  preferences?: {
+    showComments?: boolean;
+    showMentions?: boolean;
+    showRecommends?: boolean;
+  } | null;
+  shareState: ScheduledPublishData["shareState"];
+  bskyText?: string;
+  bskyFacets?: AppBskyRichtextFacet.Main[];
+  publicationUrl?: string;
+}): Promise<Result<{ scheduled_publish_at: string }, SchedulePostError>> {
+  let identity = await getIdentityData();
+  if (!identity || !identity.atp_did) return Err({ type: "not_authenticated" });
+
+  if (!identity.entitlements?.can_schedule_posts) {
+    return Err({ type: "not_found" });
+  }
+
+  if (!identity.entitlements?.publication_analytics) {
+    return Err({ type: "not_pro" });
+  }
+
+  const scheduledAt = new Date(args.scheduled_publish_at);
+  if (Number.isNaN(scheduledAt.getTime()) || scheduledAt <= new Date()) {
+    return Err({ type: "invalid_schedule" });
+  }
+
+  const data: ScheduledPublishData = {
+    shareState: args.shareState,
+    bskyText: args.bskyText,
+    bskyFacets: args.bskyFacets,
+    did: identity.atp_did,
+    publicationUrl: args.publicationUrl,
+  };
+
+  const updates: ScheduleUpdates = {
+    scheduled_publish_at: scheduledAt.toISOString(),
+    scheduled_publish_data: data as unknown as Json,
+  };
+  if (args.title !== undefined) updates.title = args.title;
+  if (args.description !== undefined) updates.description = args.description;
+  if (args.tags !== undefined) updates.tags = args.tags;
+  if (args.cover_image !== undefined) updates.cover_image = args.cover_image;
+  if (args.preferences !== undefined)
+    updates.preferences = args.preferences as unknown as Json;
+
+  if (args.publication_uri) {
+    const { data: pub } = await supabaseServerClient
+      .from("publications")
+      .select("identity_did")
+      .eq("uri", args.publication_uri)
+      .single();
+    if (!pub || pub.identity_did !== identity.atp_did) {
+      return Err({ type: "not_found" });
+    }
+  } else {
+    const { data: ownership } = await supabaseServerClient
+      .from("permission_token_on_homepage")
+      .select("token")
+      .eq("token", args.leaflet_id)
+      .eq("identity", identity.id)
+      .maybeSingle();
+    if (!ownership) return Err({ type: "not_found" });
+  }
+
+  const { found } = await updateScheduleColumns(
+    args.leaflet_id,
+    args.publication_uri,
+    updates,
+  );
+  if (!found) return Err({ type: "not_found" });
+
+  try {
+    await inngest.send({
+      // Dedupe duplicate dispatches for the same (leaflet, target time) — e.g.
+      // double-clicks or re-saving an Edit-schedule with the same value.
+      id: `scheduled-publish:${args.leaflet_id}:${scheduledAt.toISOString()}`,
+      name: "post/scheduled-publish",
+      data: {
+        leaflet_id: args.leaflet_id,
+        publication_uri: args.publication_uri,
+      },
+    });
+  } catch (e) {
+    console.log(e);
+  }
+
+  return Ok({ scheduled_publish_at: scheduledAt.toISOString() });
+}

actions/scheduledPublishDb.ts

@@ -0,0 +1,79 @@
+import { supabaseServerClient } from "supabase/serverClient";
+import { Json } from "supabase/database.types";
+
+export type ScheduleUpdates = {
+  scheduled_publish_at: string | null;
+  scheduled_publish_data: Json | null;
+  title?: string;
+  description?: string;
+  tags?: string[];
+  cover_image?: string | null;
+  preferences?: Json;
+};
+
+export async function updateScheduleColumns(
+  leaflet_id: string,
+  publication_uri: string | undefined,
+  updates: ScheduleUpdates,
+): Promise<{ found: boolean }> {
+  if (publication_uri) {
+    const { data } = await supabaseServerClient
+      .from("leaflets_in_publications")
+      .update(updates)
+      .eq("leaflet", leaflet_id)
+      .eq("publication", publication_uri)
+      .select("leaflet");
+    return { found: !!data && data.length > 0 };
+  }
+  const { data } = await supabaseServerClient
+    .from("leaflets_to_documents")
+    .update(updates)
+    .eq("leaflet", leaflet_id)
+    .select("leaflet");
+  return { found: !!data && data.length > 0 };
+}
+
+const SCHEDULE_COLUMNS =
+  "scheduled_publish_at, scheduled_publish_data, title, description, tags, cover_image, preferences, permission_tokens(root_entity)";
+
+export async function loadScheduleRow(
+  leaflet_id: string,
+  publication_uri: string | undefined,
+) {
+  if (publication_uri) {
+    const { data } = await supabaseServerClient
+      .from("leaflets_in_publications")
+      .select(SCHEDULE_COLUMNS)
+      .eq("leaflet", leaflet_id)
+      .eq("publication", publication_uri)
+      .maybeSingle();
+    return data;
+  }
+  const { data } = await supabaseServerClient
+    .from("leaflets_to_documents")
+    .select(SCHEDULE_COLUMNS)
+    .eq("leaflet", leaflet_id)
+    .maybeSingle();
+  return data;
+}
+
+export async function loadScheduledAt(
+  leaflet_id: string,
+  publication_uri: string | undefined,
+): Promise<string | null> {
+  if (publication_uri) {
+    const { data } = await supabaseServerClient
+      .from("leaflets_in_publications")
+      .select("scheduled_publish_at")
+      .eq("leaflet", leaflet_id)
+      .eq("publication", publication_uri)
+      .maybeSingle();
+    return data?.scheduled_publish_at ?? null;
+  }
+  const { data } = await supabaseServerClient
+    .from("leaflets_to_documents")
+    .select("scheduled_publish_at")
+    .eq("leaflet", leaflet_id)
+    .maybeSingle();
+  return data?.scheduled_publish_at ?? null;
+}