feat: Add support for scheduled posts

api/paidAction/itemCreate.js

@@ -3,6 +3,7 @@ import { notifyItemMention, notifyItemParents, notifyMention, notifyTerritorySub
 import { getItemMentions, getMentions, performBotBehavior } from './lib/item'
 import { msatsToSats, satsToMsats } from '@/lib/format'
 import { GqlInputError } from '@/lib/error'
+import { getScheduleAt } from '@/lib/item'
 
 export const anonable = true
 
@@ -96,6 +97,10 @@ export async function perform (args, context) {
   const mentions = await getMentions(args, context)
   const itemMentions = await getItemMentions(args, context)
 
+  // Check if this is a scheduled post
+  const scheduleAt = getScheduleAt(data.text)
+  const isScheduled = !!scheduleAt
+
   // start with median vote
   if (me) {
     const [row] = await tx.$queryRaw`SELECT
@@ -112,6 +117,8 @@ export async function perform (args, context) {
     ...data,
     ...invoiceData,
     boost,
+    isScheduled,
+    scheduledAt: scheduleAt,
     threadSubscriptions: {
       createMany: {
         data: [

api/paidAction/lib/item.js

@@ -1,5 +1,5 @@
 import { USER_ID } from '@/lib/constants'
-import { deleteReminders, getDeleteAt, getRemindAt } from '@/lib/item'
+import { deleteReminders, getDeleteAt, getRemindAt, getScheduleAt } from '@/lib/item'
 import { parseInternalLinks } from '@/lib/url'
 
 export async function getMentions ({ text }, { me, tx }) {
@@ -46,14 +46,19 @@ export const getItemMentions = async ({ text }, { me, tx }) => {
 }
 
 export async function performBotBehavior ({ text, id }, { me, tx }) {
-  // delete any existing deleteItem or reminder jobs for this item
+  // delete any existing deleteItem, reminder, or publishScheduledPost jobs for this item
   const userId = me?.id || USER_ID.anon
   id = Number(id)
   await tx.$queryRaw`
     DELETE FROM pgboss.job
-    WHERE name = 'deleteItem'
+    WHERE (name = 'deleteItem' OR name = 'publishScheduledPost')
     AND data->>'id' = ${id}::TEXT
     AND state <> 'completed'`
+  await tx.$queryRaw`
+    DELETE FROM pgboss.job
+    WHERE name = 'publishScheduledPost'
+    AND data->>'itemId' = ${id}::TEXT
+    AND state <> 'completed'`
   await deleteReminders({ id, userId, models: tx })
 
   if (text) {
@@ -85,5 +90,30 @@ export async function performBotBehavior ({ text, id }, { me, tx }) {
         }
       })
     }
+
+    const scheduleAt = getScheduleAt(text)
+    if (scheduleAt) {
+      // For new items, scheduling info is set during creation
+      // For updates, we need to update the item
+      const existingItem = await tx.item.findUnique({ where: { id: Number(id) } })
+      if (existingItem && !existingItem.isScheduled) {
+        await tx.item.update({
+          where: { id: Number(id) },
+          data: {
+            isScheduled: true,
+            scheduledAt: scheduleAt
+          }
+        })
+      }
+
+      // Schedule the job to publish the post
+      await tx.$queryRaw`
+        INSERT INTO pgboss.job (name, data, startafter, keepuntil)
+        VALUES (
+          'publishScheduledPost',
+          jsonb_build_object('itemId', ${id}::INTEGER),
+          ${scheduleAt}::TIMESTAMP WITH TIME ZONE,
+          ${scheduleAt}::TIMESTAMP WITH TIME ZONE + interval '1 minute')`
+    }
   }
 }

api/resolvers/item.js

@@ -108,7 +108,8 @@ export async function getItem (parent, { id }, { me, models }) {
       FROM "Item"
       ${whereClause(
         '"Item".id = $1',
-        activeOrMine(me)
+        activeOrMine(me),
+        scheduledOrMine(me)
       )}`
   }, Number(id))
   return item
@@ -130,6 +131,7 @@ export async function getAd (parent, { sub, subArr = [], showNsfw = false }, { m
         '"Item".bio = false',
         '"Item".boost > 0',
         activeOrMine(),
+        scheduledOrMine(me),
         subClause(sub, 1, 'Item', me, showNsfw),
         muteClause(me))}
       ORDER BY boost desc, "Item".created_at ASC
@@ -256,6 +258,12 @@ export const activeOrMine = (me) => {
 export const muteClause = me =>
   me ? `NOT EXISTS (SELECT 1 FROM "Mute" WHERE "Mute"."muterId" = ${me.id} AND "Mute"."mutedId" = "Item"."userId")` : ''
 
+export const scheduledOrMine = (me) => {
+  return me
+    ? `("Item"."isScheduled" = false OR "Item"."userId" = ${me.id})`
+    : '"Item"."isScheduled" = false'
+}
+
 const HIDE_NSFW_CLAUSE = '("Sub"."nsfw" = FALSE OR "Sub"."nsfw" IS NULL)'
 
 export const nsfwClause = showNsfw => showNsfw ? '' : HIDE_NSFW_CLAUSE
@@ -411,6 +419,7 @@ export default {
               ${whereClause(
                 `"${table}"."userId" = $3`,
                 activeOrMine(me),
+                scheduledOrMine(me),
                 nsfwClause(showNsfw),
                 typeClause(type),
                 by === 'boost' && '"Item".boost > 0',
@@ -433,6 +442,7 @@ export default {
                 '"Item"."deletedAt" IS NULL',
                 subClause(sub, 4, subClauseTable(type), me, showNsfw),
                 activeOrMine(me),
+                scheduledOrMine(me),
                 await filterClause(me, models, type),
                 typeClause(type),
                 muteClause(me)
@@ -457,6 +467,7 @@ export default {
                 typeClause(type),
                 whenClause(when, 'Item'),
                 activeOrMine(me),
+                scheduledOrMine(me),
                 await filterClause(me, models, type),
                 by === 'boost' && '"Item".boost > 0',
                 muteClause(me))}
@@ -484,6 +495,7 @@ export default {
                 typeClause(type),
                 await filterClause(me, models, type),
                 activeOrMine(me),
+                scheduledOrMine(me),
                 muteClause(me))}
               ${orderByClause('random', me, models, type)}
               OFFSET $1
@@ -513,6 +525,7 @@ export default {
                       '"parentId" IS NULL',
                       '"Item"."deletedAt" IS NULL',
                       activeOrMine(me),
+                      scheduledOrMine(me),
                       'created_at <= $1',
                       '"pinId" IS NULL',
                       subClause(sub, 4)
@@ -543,6 +556,7 @@ export default {
                         '"pinId" IS NOT NULL',
                         '"parentId" IS NULL',
                         sub ? '"subName" = $1' : '"subName" IS NULL',
+                        scheduledOrMine(me),
                         muteClause(me))}
                   ) rank_filter WHERE RANK = 1
                   ORDER BY position ASC`,
@@ -569,6 +583,7 @@ export default {
                       '"Item".bio = false',
                       ad ? `"Item".id <> ${ad.id}` : '',
                       activeOrMine(me),
+                      scheduledOrMine(me),
                       await filterClause(me, models, type),
                       subClause(sub, 3, 'Item', me, showNsfw),
                       muteClause(me))}
@@ -653,6 +668,7 @@ export default {
           ${SELECT}
           FROM "Item"
           WHERE url ~* $1 AND ("Item"."invoiceActionState" IS NULL OR "Item"."invoiceActionState" = 'PAID')
+          AND (${scheduledOrMine(me)})
           ORDER BY created_at DESC
           LIMIT 3`
       }, similar)
@@ -733,6 +749,36 @@ export default {
         homeMaxBoost: homeAgg._max.boost || 0,
         subMaxBoost: subAgg?._max.boost || 0
       }
+    },
+    scheduledItems: async (parent, { cursor, limit = LIMIT }, { me, models }) => {
+      if (!me) {
+        throw new GqlAuthenticationError()
+      }
+
+      const decodedCursor = decodeCursor(cursor)
+
+      const items = await itemQueryWithMeta({
+        me,
+        models,
+        query: `
+          ${SELECT}, trim(both ' ' from
+            coalesce(ltree2text(subpath("path", 0, -1)), '')) AS "ancestorTitles"
+          FROM "Item"
+          WHERE "userId" = $1 AND "isScheduled" = true AND "deletedAt" IS NULL
+          AND ("invoiceActionState" IS NULL OR "invoiceActionState" = 'PAID')
+          AND created_at <= $2::timestamp
+          ORDER BY "scheduledAt" ASC
+          OFFSET $3
+          LIMIT $4`,
+        orderBy: ''
+      }, me.id, decodedCursor.time, decodedCursor.offset, limit)
+
+      return {
+        cursor: items.length === limit ? nextCursorEncoded(decodedCursor, limit) : null,
+        items,
+        pins: [],
+        ad: null
+      }
     }
   },
 
@@ -1048,6 +1094,94 @@ export default {
         ])
 
       return result
+    },
+    cancelScheduledPost: async (parent, { id }, { me, models }) => {
+      if (!me) {
+        throw new GqlAuthenticationError()
+      }
+
+      const item = await models.item.findUnique({
+        where: { id: Number(id) }
+      })
+
+      if (!item) {
+        throw new GqlInputError('item not found')
+      }
+
+      if (Number(item.userId) !== Number(me.id)) {
+        throw new GqlInputError('item does not belong to you')
+      }
+
+      if (!item.isScheduled) {
+        throw new GqlInputError('item is not scheduled')
+      }
+
+      // Cancel the scheduled job
+      await models.$queryRaw`
+        DELETE FROM pgboss.job
+        WHERE name = 'publishScheduledPost'
+        AND data->>'itemId' = ${item.id}::TEXT
+        AND state <> 'completed'`
+
+      // Update the item to remove scheduling
+      return await models.item.update({
+        where: { id: Number(id) },
+        data: {
+          isScheduled: false,
+          scheduledAt: null
+        }
+      })
+    },
+    publishScheduledPostNow: async (parent, { id }, { me, models }) => {
+      if (!me) {
+        throw new GqlAuthenticationError()
+      }
+
+      const item = await models.item.findUnique({
+        where: { id: Number(id) }
+      })
+
+      if (!item) {
+        throw new GqlInputError('item not found')
+      }
+
+      if (Number(item.userId) !== Number(me.id)) {
+        throw new GqlInputError('item does not belong to you')
+      }
+
+      if (!item.isScheduled) {
+        throw new GqlInputError('item is not scheduled')
+      }
+
+      // Cancel the scheduled job
+      await models.$queryRaw`
+    DELETE FROM pgboss.job
+    WHERE name = 'publishScheduledPost'
+    AND data->>'itemId' = ${item.id}::TEXT
+    AND state <> 'completed'`
+
+      const publishTime = new Date()
+
+      // Publish immediately with current timestamp
+      const updatedItem = await models.item.update({
+        where: { id: Number(id) },
+        data: {
+          isScheduled: false,
+          scheduledAt: null,
+          createdAt: publishTime,
+          updatedAt: publishTime
+        }
+      })
+
+      // Refresh cached views
+      await models.$executeRaw`REFRESH MATERIALIZED VIEW CONCURRENTLY hot_score_view`
+
+      // Queue side effects
+      await models.$executeRaw`
+    INSERT INTO pgboss.job (name, data, startafter)
+    VALUES ('schedulePostSideEffects', jsonb_build_object('itemId', ${item.id}::INTEGER), now())`
+
+      return updatedItem
     }
   },
   ItemAct: {
@@ -1357,7 +1491,8 @@ export default {
           FROM "Item"
           ${whereClause(
             '"Item".id = $1',
-            `("Item"."invoiceActionState" IS NULL OR "Item"."invoiceActionState" = 'PAID'${me ? ` OR "Item"."userId" = ${me.id}` : ''})`
+            `("Item"."invoiceActionState" IS NULL OR "Item"."invoiceActionState" = 'PAID'${me ? ` OR "Item"."userId" = ${me.id}` : ''})`,
+            scheduledOrMine(me)
           )}`
       }, Number(item.rootId))
 
@@ -1416,6 +1551,17 @@ export default {
         AND data->>'userId' = ${meId}::TEXT
         AND state = 'created'`
       return reminderJobs[0]?.startafter ?? null
+    },
+    scheduledAt: async (item, args, { me, models }) => {
+      const meId = me?.id ?? USER_ID.anon
+      if (meId !== item.userId) {
+        // Only show scheduledAt for your own items to keep DB queries minimized
+        return null
+      }
+      return item.scheduledAt
+    },
+    isScheduled: async (item, args, { me, models }) => {
+      return !!item.isScheduled
     }
   }
 }

api/resolvers/search.js

@@ -1,6 +1,6 @@
 import { decodeCursor, LIMIT, nextCursorEncoded } from '@/lib/cursor'
 import { whenToFrom } from '@/lib/time'
-import { getItem, itemQueryWithMeta, SELECT } from './item'
+import { getItem, itemQueryWithMeta, SELECT, scheduledOrMine } from './item'
 import { parse } from 'tldts'
 
 function queryParts (q) {
@@ -163,7 +163,8 @@ export default {
           WITH r(id, rank) AS (VALUES ${values})
           ${SELECT}, rank
           FROM "Item"
-          JOIN r ON "Item".id = r.id`,
+          JOIN r ON "Item".id = r.id
+          WHERE ${scheduledOrMine(me)}`,
         orderBy: 'ORDER BY rank ASC'
       })
 
@@ -501,7 +502,8 @@ export default {
           WITH r(id, rank) AS (VALUES ${values})
           ${SELECT}, rank
           FROM "Item"
-          JOIN r ON "Item".id = r.id`,
+          JOIN r ON "Item".id = r.id
+          WHERE ${scheduledOrMine(me)}`,
         orderBy: 'ORDER BY rank ASC, msats DESC'
       })).map((item, i) => {
         const e = sitems.body.hits.hits[i]

api/typeDefs/item.js

@@ -11,6 +11,7 @@ export default gql`
     auctionPosition(sub: String, id: ID, boost: Int): Int!
     boostPosition(sub: String, id: ID, boost: Int): BoostPositions!
     itemRepetition(parentId: ID): Int!
+    scheduledItems(cursor: String, limit: Limit): Items
   }
 
   type BoostPositions {
@@ -63,6 +64,8 @@ export default gql`
     act(id: ID!, sats: Int, act: String, hasSendWallet: Boolean): ItemActPaidAction!
     pollVote(id: ID!): PollVotePaidAction!
     toggleOutlaw(id: ID!): Item!
+    cancelScheduledPost(id: ID!): Item!
+    publishScheduledPostNow(id: ID!): Item!
   }
 
   type PollVoteResult {
@@ -112,6 +115,8 @@ export default gql`
     deletedAt: Date
     deleteScheduledAt: Date
     reminderScheduledAt: Date
+    scheduledAt: Date
+    isScheduled: Boolean!
     title: String
     searchTitle: String
     url: String